Ticket #295: remotepostsource.patch

xapian-core/matcher/externalpostlist.cc

@@ -29 +29 @@
 
 using namespace std;
 
-ExternalPostList::ExternalPostList(Xapian::PostingSource *source_,
+ExternalPostList::ExternalPostList(const Xapian::Database & db,
+                                   Xapian::PostingSource *source_,
                                    double factor_)
-    : source(source_), current(0), factor(factor_)
+    : source(source_), source_is_owned(false), current(0), factor(factor_)
 {
     Assert(source);
-    source->reset();
+    Xapian::PostingSource * newsource = source->clone();
+    if (newsource != NULL) {
+        source = newsource;
+        source_is_owned = true;
+    }
+    source->reset(db);
 }
 
+ExternalPostList::~ExternalPostList()
+{
+    if (source_is_owned) {
+        delete source;
+    }
+}
+
 Xapian::doccount
 ExternalPostList::get_termfreq_min() const
 {

xapian-core/matcher/queryoptimiser.cc

@@ -64 +64 @@
 
         case Xapian::Query::Internal::OP_EXTERNAL_SOURCE:
             Assert(query->external_source);
-            RETURN(new ExternalPostList(query->external_source, factor));
+            RETURN(new ExternalPostList(
+                Xapian::Database(const_cast<Xapian::Database::Internal *>(&db)),
+                query->external_source, factor));
 
         case Xapian::Query::OP_AND:
         case Xapian::Query::OP_FILTER:

xapian-core/matcher/externalpostlist.h

@@ -29 +29 @@
 
 class ExternalPostList : public PostList {
     Xapian::PostingSource * source;
+    bool source_is_owned;
 
     Xapian::docid current;
 
@@ -43 +44 @@
     PostList * update_after_advance();
 
   public:
-    ExternalPostList(Xapian::PostingSource *source_, double factor_);
+    ExternalPostList(const Xapian::Database & db,
+                     Xapian::PostingSource *source_,
+                     double factor_);
+    ~ExternalPostList();
 
     Xapian::doccount get_termfreq_min() const;
 

xapian-core/tests/api_db.cc

@@ -1887 +1887 @@
         : num_docs(db.get_doccount()), last_docid(db.get_lastdocid()), did(0)
     { }
 
-    void reset() { did = 0; }
+    MyOddPostingSource(Xapian::doccount num_docs_,
+                       Xapian::doccount last_docid_)
+        : num_docs(num_docs_), last_docid(last_docid_), did(0)
+    { }
 
+    PostingSource * clone() const { return new MyOddPostingSource(num_docs, last_docid); }
+
+    void reset(const Xapian::Database &) { did = 0; }
+
     // These bounds could be better, but that's not important here.
     Xapian::doccount get_termfreq_min() const { return 0; }
 
@@ -1925 +1932 @@
     Xapian::Enquire enq(db);
     MyOddPostingSource src(db);
 
-    // Check that passing NULL is rejected as intended.
-    TEST_EXCEPTION(Xapian::InvalidArgumentError, Xapian::Query bad(NULL));
-    Xapian::PostingSource * nullsrc = NULL;
-    TEST_EXCEPTION(Xapian::InvalidArgumentError, Xapian::Query bad(nullsrc));
-                
-    enq.set_query(Xapian::Query(&src));
+    enq.set_query(Xapian::Query(src));
 
     Xapian::MSet mset = enq.get_mset(0, 10);
     mset_expect_order(mset, 1, 3, 5, 7, 9, 11, 13, 15, 17);
 
     Xapian::Query q(Xapian::Query::OP_FILTER,
                     Xapian::Query("leav"),
-                    Xapian::Query(&src));
+                    Xapian::Query(src));
     enq.set_query(q);
 
     mset = enq.get_mset(0, 10);
@@ -1953 +1955 @@
     Xapian::Enquire enq(db);
     MyOddPostingSource src(db);
 
-    enq.set_query(Xapian::Query(&src));
+    enq.set_query(Xapian::Query(src));
 
     TEST_EXCEPTION(Xapian::UnimplementedError,
                    Xapian::MSet mset = enq.get_mset(0, 10));
 
     Xapian::Query q(Xapian::Query::OP_FILTER,
                     Xapian::Query("leav"),
-                    Xapian::Query(&src));
+                    Xapian::Query(src));
     enq.set_query(q);
 
     TEST_EXCEPTION(Xapian::UnimplementedError,
@@ -1981 +1983 @@
         : num_docs(db.get_doccount()), last_docid(db.get_lastdocid()), did(0)
     { }
 
-    void reset() { did = 0; }
+    MyOddWeightingPostingSource(Xapian::doccount num_docs_,
+                                Xapian::doccount last_docid_)
+        : num_docs(num_docs_), last_docid(last_docid_), did(0)
+    { }
 
+    PostingSource * clone() const { return new MyOddWeightingPostingSource(num_docs, last_docid); }
+
+    void reset(const Xapian::Database &) { did = 0; }
+
     Xapian::weight get_weight() const {
         return (did % 2) ? 1000 : 0.001;
     }
@@ -2028 +2037 @@
     Xapian::Enquire enq(db);
     MyOddWeightingPostingSource src(db);
 
-    enq.set_query(Xapian::Query(&src));
+    enq.set_query(Xapian::Query(src));
 
     Xapian::MSet mset = enq.get_mset(0, 10);
     mset_expect_order(mset, 1, 3, 5, 7, 9, 11, 13, 15, 17, 2);
 
     Xapian::Query q(Xapian::Query::OP_OR,
                     Xapian::Query("leav"),
-                    Xapian::Query(&src));
+                    Xapian::Query(src));
     enq.set_query(q);
 
     mset = enq.get_mset(0, 5);
@@ -2051 +2060 @@
     tout << "max possible weight = " << mset.get_max_possible() << endl;
     TEST(mset.get_max_possible() > 1000);
 
-    enq.set_query(Xapian::Query(q.OP_SCALE_WEIGHT, Xapian::Query(&src), 0.5));
+    enq.set_query(Xapian::Query(q.OP_SCALE_WEIGHT, Xapian::Query(src), 0.5));
     mset = enq.get_mset(0, 10);
     TEST(mset.empty());
 
     TEST_EQUAL(mset.get_max_possible(), 500);
 
-    enq.set_query(Xapian::Query(q.OP_SCALE_WEIGHT, Xapian::Query(&src), 2));
+    enq.set_query(Xapian::Query(q.OP_SCALE_WEIGHT, Xapian::Query(src), 2));
     mset = enq.get_mset(0, 10);
     mset_expect_order(mset, 1, 3, 5, 7, 9, 11, 13, 15, 17);
 
@@ -2078 +2087 @@
         : num_docs(db.get_doccount()), last_docid(db.get_lastdocid()), did(0)
     { }
 
-    void reset() { did = 0; }
+    MyDontAskWeightPostingSource(Xapian::doccount num_docs_,
+                                 Xapian::doccount last_docid_)
+        : num_docs(num_docs_), last_docid(last_docid_), did(0)
+    { }
 
+    PostingSource * clone() const { return new MyDontAskWeightPostingSource(num_docs, last_docid); }
+
+    void reset(const Xapian::Database &) { did = 0; }
+
     Xapian::weight get_weight() const {
         FAIL_TEST("MyDontAskWeightPostingSource::get_weight() called");
     }
@@ -2126 +2142 @@
     MyDontAskWeightPostingSource src(db);
 
     tout << "OP_SCALE_WEIGHT 0" << endl;
-    enq.set_query(Xapian::Query(Xapian::Query::OP_SCALE_WEIGHT, Xapian::Query(&src), 0));
+    enq.set_query(Xapian::Query(Xapian::Query::OP_SCALE_WEIGHT, Xapian::Query(src), 0));
 
     Xapian::MSet mset = enq.get_mset(0, 5);
     mset_expect_order(mset, 1, 2, 3, 4, 5);
@@ -2134 +2150 @@
     tout << "OP_FILTER" << endl;
     Xapian::Query q(Xapian::Query::OP_FILTER,
                     Xapian::Query("leav"),
-                    Xapian::Query(&src));
+                    Xapian::Query(src));
     enq.set_query(q);
 
     mset = enq.get_mset(0, 5);
     mset_expect_order(mset, 8, 6, 4, 5, 7);
 
     tout << "BoolWeight" << endl;
-    enq.set_query(Xapian::Query(&src));
+    enq.set_query(Xapian::Query(src));
     enq.set_weighting_scheme(Xapian::BoolWeight());
 
     //mset = enq.get_mset(0, 5);
@@ -2151 +2167 @@
 }
 
 // Check that valueweightsource works correctly.
-DEFINE_TESTCASE(valueweightsource1, backend && !remote) {
-    // FIXME: PostingSource doesn't currently work well with multi databases
-    // but we should try to resolve that issue.
-    SKIP_TEST_FOR_BACKEND("multi");
+DEFINE_TESTCASE(valueweightsource1, backend) {
     Xapian::Database db(get_database("apitest_phrase"));
     Xapian::Enquire enq(db);
-    Xapian::ValueWeightPostingSource src(db, 11);
+    Xapian::ValueWeightPostingSource src(11);
 
     // Should be in descending order of length
     tout << "RAW" << endl;
-    enq.set_query(Xapian::Query(&src));
+    enq.set_query(Xapian::Query(src));
     Xapian::MSet mset = enq.get_mset(0, 5);
     mset_expect_order(mset, 3, 1, 2, 8, 14);
 
@@ -2169 +2182 @@
     tout << "OP_FILTER" << endl;
     Xapian::Query q(Xapian::Query::OP_FILTER,
                     Xapian::Query("leav"),
-                    Xapian::Query(&src));
+                    Xapian::Query(src));
     enq.set_query(q);
     mset = enq.get_mset(0, 5);
     mset_expect_order(mset, 8, 6, 4, 5, 7);
@@ -2177 +2190 @@
     // Should be in descending order of length
     tout << "OP_FILTER other way" << endl;
     q = Xapian::Query(Xapian::Query::OP_FILTER,
-                      Xapian::Query(&src),
+                      Xapian::Query(src),
                       Xapian::Query("leav"));
     enq.set_query(q);
     mset = enq.get_mset(0, 5);
@@ -2189 +2202 @@
 // Check that valueweightsource gives the correct bounds for those databases
 // which support value statistics.
 DEFINE_TESTCASE(valueweightsource2, backend && valuestats) {
-    // FIXME: PostingSource doesn't currently work well with multi databases
-    // but we should try to resolve that issue.
-    SKIP_TEST_FOR_BACKEND("multi");
     Xapian::Database db(get_database("apitest_phrase"));
-    Xapian::ValueWeightPostingSource src(db, 11);
+    Xapian::ValueWeightPostingSource src(11);
+    src.reset(db);
     TEST_EQUAL(src.get_termfreq_min(), 17);
     TEST_EQUAL(src.get_termfreq_est(), 17);
     TEST_EQUAL(src.get_termfreq_max(), 17);
@@ -2204 +2215 @@
 
 // Check that valueweightsource skip_to() can stay in the same position.
 DEFINE_TESTCASE(valueweightsource3, backend && valuestats) {
-    // FIXME: PostingSource doesn't currently work well with multi databases
-    // but we should try to resolve that issue.
-    SKIP_TEST_FOR_BACKEND("multi");
     Xapian::Database db(get_database("apitest_phrase"));
-    Xapian::ValueWeightPostingSource src(db, 11);
+    Xapian::ValueWeightPostingSource src(11);
+    src.reset(db);
     TEST(!src.at_end());
     src.skip_to(8, 0.0);
     TEST(!src.at_end());

xapian-core/tests/api_percentages.cc

@@ -45 +45 @@
         : maxwt(0.0), started(false)
     {}
 
+    MyPostingSource(const std::vector<std::pair<Xapian::docid, Xapian::weight> > &weights_,
+                    Xapian::weight maxwt_)
+        : weights(weights_), maxwt(maxwt_), started(false)
+    {}
+
+
+    PostingSource * clone() const
+    {
+        return new MyPostingSource(weights, maxwt);
+    }
+
     void append_docweight(Xapian::docid did, Xapian::weight wt)
     {
         weights.push_back(make_pair(did, wt));
@@ -55 +66 @@
         if (wt > maxwt) maxwt = wt;
     }
 
-    void reset() { started = false; }
+    void reset(const Xapian::Database &) { started = false; }
 
     Xapian::weight get_weight() const {
         return i->second;
@@ -103 +114 @@
         MyPostingSource source;
         source.append_docweight(1, 100);
         source.append_docweight(2, 50 - epsilons * DBL_EPSILON);
-        enquire.set_query(Xapian::Query(&source));
+        enquire.set_query(Xapian::Query(source));
         Xapian::MSet mset = enquire.get_mset(0, 10);
         TEST_EQUAL(mset.size(), 2);
         if (mset[1].get_percent() != 50) break;
@@ -120 +131 @@
         source.append_docweight(4, 50 - epsilons * DBL_EPSILON);
         source.append_docweight(5, 25);
 
-        enquire.set_query(Xapian::Query(&source));
+        enquire.set_query(Xapian::Query(source));
         Xapian::MSet mset1 = enquire.get_mset(0, 10);
         TEST_EQUAL(mset1.size(), 5);
         TEST_EQUAL(mset1[2].get_percent(), 50);

xapian-core/include/xapian/postingsource.h

@@ -137 +137 @@
     /// Return the current docid.
     virtual Xapian::docid get_docid() const = 0;
 
+    /** Clone the posting source.
+     *
+     *  The clone should inherit the configuration of the parent, but need not
+     *  inherit the state.  ie, the clone does not need to be in the same
+     *  iteration position as the original: the matcher will always call
+     *  reset() on the clone before attempting to move the iterator, or read
+     *  the information about the current position of the iterator.
+     */
+    virtual PostingSource * clone() const = 0;
+
+    /** Name of the posting source, for performing remote searches.
+     *
+     *  If the subclass is called FooPostingSource, this should return "Foo".
+     *
+     *  This should only be implemented if serialise() and unserialise() are
+     *  also implemented.
+     *
+     *  The default implmenentation returns an empty string, to indicate that
+     *  serialise() and unserialise() are not implemented.
+     */
+    virtual std::string name() const;
+
+    /** Serialise object parameters into a string.
+     *
+     *  The serialised parameters should represent the configuration of the
+     *  posting source, but need not (indeed, should not) represent the current
+     *  iteration state.
+     */
+    virtual std::string serialise() const;
+
+    /** Create object given string serialisation returned by serialise().
+     *
+     *  @param s A serialised instance of this PostingSource subclass.
+     */
+    virtual PostingSource * unserialise(const std::string &s) const;
+
     /** Reset this PostingSource to its freshly constructed state.
      *
      *  This is called automatically by the matcher prior to each query being
      *  processed.
+     *
+     *  FIXME - document the db parameter.
      */
-    virtual void reset() = 0;
+    virtual void reset(const Database & db) = 0;
 
     /** Return a string describing this object.
      *
@@ -196 +234 @@
     /// An upper bound on the value returned.
     double max_value;
 
+    /// Upper bound on the value returned specified in constructor.
+    double specified_max_value;
+
   public:
     /** Construct a ValueWeightPostingSource.
      *
      *  @param db_ The database to read values from.
      *  @param valno_ The value slot to read values from.
      */
-    ValueWeightPostingSource(Xapian::Database db_, Xapian::valueno valno_);
+    ValueWeightPostingSource(Xapian::valueno valno_);
 
     /** Construct a ValueWeightPostingSource.
      *
@@ -214 +255 @@
      *  constructor need only be used if more accurate information is
      *  available.
      */
-    ValueWeightPostingSource(Xapian::Database db_, Xapian::valueno valno_,
-                             double max_weight_);
+    ValueWeightPostingSource(Xapian::valueno valno_, double max_weight_);
 
     Xapian::doccount get_termfreq_min() const;
     Xapian::doccount get_termfreq_est() const;
@@ -232 +272 @@
 
     Xapian::docid get_docid() const;
 
-    void reset();
+    ValueWeightPostingSource * clone() const;
+    std::string name() const;
+    std::string serialise() const;
+    PostingSource * unserialise(const std::string &s) const;
 
+    void reset(const Database & db_);
+
     std::string get_description() const;
 };
 

xapian-core/include/xapian/query.h

@@ -25 +25 @@
 #ifndef XAPIAN_INCLUDED_QUERY_H
 #define XAPIAN_INCLUDED_QUERY_H
 
+#include <map>
 #include <string>
 #include <vector>
 
@@ -204 +205 @@
          */
         Query(Query::op op_, Xapian::valueno valno, const std::string &value);
 
-        /// Construct an external source query.
-        explicit Query(Xapian::PostingSource * external_source);
+        /** Construct an external source query.
+         *
+         *  The posting source will be cloned immediately, so the source
+         *  supplied may be safely deallocated after this call.
+         *
+         *  @param external_source The source to use in the query.
+         */
+        explicit Query(const Xapian::PostingSource & external_source);
 
         /** A query which matches all documents in the database. */
         static Xapian::Query MatchAll;
@@ -405 +412 @@
         /** Destructor. */
         ~Internal();
 
-        static Xapian::Query::Internal * unserialise(const std::string &s);
+        static Xapian::Query::Internal * unserialise(const std::string &s,
+                const std::map<std::string, Xapian::PostingSource *> &sources);
 
         /** Add a subquery.
          */

xapian-core/net/remoteserver.cc

@@ -114 +114 @@
     wtschemes[weight->name()] = weight;
     weight = new Xapian::TradWeight;
     wtschemes[weight->name()] = weight;
+
+    Xapian::PostingSource * source;
+    source = new Xapian::ValueWeightPostingSource(0);
+    postingsources[source->name()] = source;
 }
 
 RemoteServer::~RemoteServer()
@@ -121 +125 @@
     delete db;
     // wdb is either NULL or equal to db, so we shouldn't delete it too!
 
-    map<string, Xapian::Weight*>::const_iterator i;
-    for (i = wtschemes.begin(); i != wtschemes.end(); ++i) {
-        delete i->second;
+    {
+        map<string, Xapian::Weight*>::const_iterator i;
+        for (i = wtschemes.begin(); i != wtschemes.end(); ++i) {
+            delete i->second;
+        }
     }
+
+    {
+        map<string, Xapian::PostingSource *>::const_iterator i;
+        for (i = postingsources.begin(); i != postingsources.end(); ++i) {
+            delete i->second;
+        }
+    }
 }
 
 message_type
@@ -365 +378 @@
 
     // Unserialise the Query.
     len = decode_length(&p, p_end, true);
-    AutoPtr<Xapian::Query::Internal> query(Xapian::Query::Internal::unserialise(string(p, len)));
+    AutoPtr<Xapian::Query::Internal> query(Xapian::Query::Internal::unserialise(string(p, len), postingsources));
     p += len;
 
     // Unserialise assorted Enquire settings.
@@ -610 +623 @@
 
     send_message(REPLY_ADDDOCUMENT, encode_length(did));
 }
+
+
+void
+RemoteServer::register_posting_source(const Xapian::PostingSource &source)
+{
+    if (source.name().empty()) {
+        throw Xapian::InvalidOperationError("Unable to register posting source - name() method returns empty string.");
+    }
+    Xapian::PostingSource * sourceclone = source.clone();
+    if (!sourceclone) {
+        throw Xapian::InvalidOperationError("Unable to register posting source - clone() method returns NULL.");
+    }
+    try {
+        postingsources[source.name()] = sourceclone;
+    } catch(...) {
+        delete sourceclone;
+        throw;
+    }
+}

xapian-core/common/remoteserver.h

@@ -24 +24 @@
 
 #include "xapian/database.h"
 #include "xapian/enquire.h"
+#include "xapian/postingsource.h"
 #include "xapian/visibility.h"
 
 #include "remoteconnection.h"
@@ -31 +32 @@
 #include <map>
 #include <string>
 
-// Forward declaration
-namespace Xapian { class Weight; }
-
 using namespace std;
 
 /** Remote backend server base class. */
@@ -70 +68 @@
     /// Registered weighting schemes.
     map<string, Xapian::Weight *> wtschemes;
 
+    /// Registered external posting sources.
+    map<string, Xapian::PostingSource *> postingsources;
+
     /// Accept a message from the client.
     message_type get_message(Xapian::timeout timeout, string & result,
                              message_type required_type = MSG_MAX);
@@ -173 +174 @@
     void register_weighting_scheme(const Xapian::Weight &wt) {
         wtschemes[wt.name()] = wt.clone();
     }
+
+    /** Register a user-defined posting source class.
+     */
+    void register_posting_source(const Xapian::PostingSource &source);
 };
 
 #endif // XAPIAN_INCLUDED_REMOTESERVER_H

xapian-core/api/omqueryinternal.cc

@@ -42 +42 @@
 #include <cfloat>
 #include <climits>
 #include <cmath>
+#include <map>
 #include <set>
 #include <vector>
 
@@ -142 +143 @@
         result += '[';
         result += encode_length(tname.length());
         result += tname;
-        if (term_pos != curpos) result += '@' + om_tostring(term_pos);
-        if (wqf != 1) result += '#' + om_tostring(wqf);
+        if (term_pos != curpos) result += '@' + om_tostring(term_pos); // FIXME - should this use encode_length()?
+        if (wqf != 1) result += '#' + om_tostring(wqf); // FIXME - should this use encode_length()?
         ++curpos;
     } else if (op == Xapian::Query::Internal::OP_EXTERNAL_SOURCE) {
-        throw Xapian::UnimplementedError("Remote backend doesn't support PostingSource");
+        string sourcename = external_source->name();
+        if (sourcename.empty())
+            throw Xapian::UnimplementedError("This PostingSource doesn't support remote use.");
+        result += '!';
+        result += encode_length(sourcename.length());
+        result += sourcename;
+        string sourcedata = external_source->serialise();
+        result += encode_length(sourcedata.length());
+        result += sourcedata;
     } else {
         result += "(";
         for (subquery_list::const_iterator i = subqs.begin();
@@ -158 +167 @@
             case Xapian::Query::Internal::OP_LEAF:
                 Assert(false);
                 break;
+            case Xapian::Query::Internal::OP_EXTERNAL_SOURCE:
+                Assert(false);
+                break;
             case Xapian::Query::OP_AND:
                 result += "&";
                 break;
@@ -380 +392 @@
     const char *p;
     const char *end;
     Xapian::termpos curpos;
+    const map<string, Xapian::PostingSource *> & sources;
 
     Xapian::Query::Internal * readquery();
+    Xapian::Query::Internal * readexternal();
     Xapian::Query::Internal * readcompound();
 
   public:
-    QUnserial(const string & s) : p(s.c_str()), end(p + s.size()), curpos(1) { }
+    QUnserial(const string & s,
+              const map<string, Xapian::PostingSource *> & sources_)
+            : p(s.c_str()), end(p + s.size()), curpos(1), sources(sources_) { }
     Xapian::Query::Internal * decode();
 };
 
@@ -424 +440 @@
             ++curpos;
             return new Xapian::Query::Internal(tname, wqf, term_pos);
         }
+        case '!':
+            return readexternal();
         case '(':
             return readcompound();
         default:
@@ -432 +450 @@
     }
 }
 
+Xapian::Query::Internal *
+QUnserial::readexternal()
+{
+    if (p == end)
+        throw Xapian::InvalidArgumentError("Bad serialised query");
+
+    size_t length = decode_length(&p, end, true);
+    string sourcename(p, length);
+    map<string, Xapian::PostingSource *>::const_iterator i;
+    i = sources.find(string(p, length));
+    if (i == sources.end()) {
+        throw Xapian::InvalidArgumentError("PostingSource " + string(p, length) + " not registered");
+    }
+
+    p += length;
+    length = decode_length(&p, end, true);
+    string sourcedata(p, length);
+    p += length;
+
+    return new Xapian::Query::Internal(i->second->unserialise(sourcedata));
+}
+
 static Xapian::Query::Internal *
 qint_from_vector(Xapian::Query::op op,
                  const vector<Xapian::Query::Internal *> & vec,
@@ -476 +516 @@
                     --p;
                     subqs.push_back(readquery());
                     break;
+                case '!':