Ticket #340: batch_processing.diff

ruby/test/unit/index/tc_index.rb

@@ -462 +462 @@
     index.close
   end
 
+  def test_index_key_batch0
+    data = {
+      "0" => {:id => "0", :val => "one"},
+      "0" => {:id => "0", :val => "two"},
+      "1" =>{:id => "1", :val => "three"},
+      "1" => {:id => "1", :val => "four"},
+    }
+
+    index = Index.new(:analyzer => WhiteSpaceAnalyzer.new,
+                      :key => :id)
+    index.batch_update data
+    assert_equal(2, index.size)
+    index.close
+  end
+
+  def test_index_key_batch1
+    data0 = {
+      "0" => {:id => "0", :val => "one"},
+      "0" => {:id => "0", :val => "two"},
+      "1" =>{:id => "1", :val => "three"},
+      "2" => {:id => "1", :val => "four"},
+    }
+
+    data1 = {
+      "0" => {:id => "0", :val => "one"},
+      "3" => {:id => "3", :val => "two"},
+      "2" =>{:id => "2", :val => "three"},
+      "1" => {:id => "1", :val => "four"},
+      "4" => {:id => "4", :val => "four"},
+    }
+
+    index = Index.new(:analyzer => WhiteSpaceAnalyzer.new,
+                      :key => :id)
+    index.batch_update data0
+    assert_equal(3, index.size)
+    index.batch_update data1
+    assert_equal(5, index.size)
+    index.close
+  end
+
+  def test_index_key_delete_batch0
+    data0 = {
+      "0" => {:id => "0", :val => "one"},
+      "0" => {:id => "0", :val => "two"},
+      "1" =>{:id => "1", :val => "three"},
+      "2" => {:id => "2", :val => "four"},
+      "0" => {:id => "0", :val => "four"},
+    }
+
+    data1 = ["0", "1"];
+
+    index = Index.new(:analyzer => WhiteSpaceAnalyzer.new, :key => :id)
+    index.batch_update data0
+
+    assert_equal("four", index["0"][:val])
+    assert_equal("three", index["1"][:val])
+    assert_equal("four", index["2"][:val])
+
+    assert_equal(3, index.size)
+    index.batch_delete data1
+    assert_equal(1, index.size)
+    assert_equal("four", index["2"][:val])
+
+    index.close
+  end
+
   def test_index_multi_key
     index = Index.new(:analyzer => WhiteSpaceAnalyzer.new,
                       :key => [:id, :table])

ruby/ext/r_index.c

@@ -1627 +1627 @@
 }
 
 /*
+ * Same as frt_iw_delete, but +rterms+ is a vector of terms.
+ */
+static VALUE
+frt_iw_batch_delete(VALUE self, VALUE rfield, VALUE rterms)
+{
+    IndexWriter *iw = (IndexWriter *) DATA_PTR(self);
+    iw_batch_delete_term(iw, frt_field(rfield), rterms);
+    return self;
+}
+
+/*
  *  call-seq:
  *     index_writer.field_infos -> FieldInfos
  *
@@ -3298 +3309 @@
     rb_define_method(cIndexWriter, "commit",        frt_iw_commit, 0);
     rb_define_method(cIndexWriter, "add_readers",   frt_iw_add_readers, 1);
     rb_define_method(cIndexWriter, "delete",        frt_iw_delete, 2);
+    rb_define_method(cIndexWriter, "batch_delete",  frt_iw_batch_delete, 2);
     rb_define_method(cIndexWriter, "field_infos",   frt_iw_field_infos, 0);
     rb_define_method(cIndexWriter, "analyzer",      frt_iw_get_analyzer, 0);
     rb_define_method(cIndexWriter, "analyzer=",     frt_iw_set_analyzer, 1);

ruby/lib/ferret/index.rb

@@ -451 +451 @@
       return self
     end
 
+    # helper function for batch_delete
+    def batch_delete_one(id, remaining) 
+      if id.is_a?(String) or id.is_a?(Symbol) 
+        remaining << id.to_s
+      elsif id.is_a?(Integer)
+        ensure_reader_open()
+        cnt = @reader.delete(id)
+      else
+        raise ArgumentError, "Cannot delete for arg of type #{id.class}"
+      end
+    end
+    
+    # unlocked version of batch_delete
+    def batch_delete_unlocked(docs)
+      remaining = []
+      if docs.is_a?(Hash)
+        docs.each do |id, docs|          
+          batch_delete_one(id, remaining)
+        end
+      elsif docs.is_a?(Array)
+        docs.each do |id|          
+          batch_delete_one(id, remaining)
+        end
+      else
+        raise ArgumentError, "Cannot batch delete for arg of type #{docs.class}"
+      end
+      if(@reader) 
+        @reader.commit
+      end
+      ensure_writer_open()
+      if remaining.length != 0
+        @writer.batch_delete(@id_field, remaining)
+      end
+      @writer.commit
+      return self
+    end
+    
+    # Deletes documents from the index. 
+    # +docs+ contains ids, which are processed according to their type.
+    # see delete(arg) for more precision
+    # docs:: An array of docs to be deleted, or a hash (key is then only used)
+    def batch_delete(docs)
+      @dir.synchrolock do
+        batch_delete_unlocked(docs)
+      end
+      return self
+    end
+
     # Delete all documents returned by the query.
     # 
     # query:: The query to find documents you wish to delete. Can either be a
@@ -479 +527 @@
     # Update the document referenced by the document number +id+ if +id+ is an
     # integer or all of the documents which have the term +id+ if +id+ is a
     # term..
+    # For batch update of set of documents, for performance reasons, see batch_update
     #
     # id::      The number of the document to update. Can also be a string
     #           representing the value in the +id+ field. Also consider using
@@ -498 +547 @@
       end
     end
 
+    # Update the documents referenced by the documents contained in +docs+ 
+    # Each entry in +ids+ is a id/doc pair
+    # See update(id, new_doc) for more precision on the way id is processed
+    # according to its type
+    # docs::    The set of documents to be updated
+    def batch_update(docs)
+      @dir.synchrolock do
+        batch_delete_unlocked(docs)
+        ensure_writer_open()
+        docs.each do |id, new_doc| 
+          @writer << new_doc
+        end
+        flush()
+      end
+    end
+
+
     # Update all the documents returned by the query.
     #
     # query::   The query to find documents you wish to update. Can either be

c/include/index.h

@@ -924 +924 @@
                             const Config *config);
 extern void iw_delete_term(IndexWriter *iw, const char *field,
                            const char *term);
+extern void iw_batch_delete_term(IndexWriter *iw, const char *field, 
+                                 VALUE terms);
 extern void iw_close(IndexWriter *iw);
 extern void iw_add_doc(IndexWriter *iw, Document *doc);
 extern int iw_doc_count(IndexWriter *iw);

c/src/index.c

@@ -6049 +6049 @@
     }
 }
 
+void iw_batch_delete_term(IndexWriter *iw, const char *field, VALUE terms)
+{
+    int field_num = fis_get_field_num(iw->fis, field);
+    if (field_num >= 0) {
+        int i;
+        mutex_lock(&iw->mutex);
+        iw_commit_i(iw);
+        do {
+            SegmentInfos *sis = iw->sis;
+            const int seg_cnt = sis->size;
+            bool did_delete = false;
+            for (i = 0; i < seg_cnt; i++) {
+                IndexReader *ir = sr_open(sis, iw->fis, i, false);
+                unsigned int i;
+                for (i = 0 ; i < RARRAY(terms)->len ; i++) {
+                    TermDocEnum *tde = ir->term_docs(ir);
+                    VALUE rterm = RARRAY(terms)->ptr[i];
+                    const char* term = StringValuePtr(rterm);
+                    ir->deleter = iw->deleter;
+                    stde_seek(tde, field_num, term);
+                    while (tde->next(tde)) {
+                        did_delete = true;
+                        sr_delete_doc_i(ir, STDE(tde)->doc_num);
+                    }
+                    tde_destroy(tde);
+                }
+                sr_commit_i(ir);
+                ir_close(ir);
+            }
+            if (did_delete) {
+                mutex_lock(&iw->store->mutex);
+                sis_write(iw->sis, iw->store, iw->deleter);
+                mutex_unlock(&iw->store->mutex);
+            }
+        } while (0);
+        mutex_unlock(&iw->mutex);
+    }
+}
+
 static void iw_optimize_i(IndexWriter *iw)
 {
     int min_segment;