Ticket #295: remotepostsource2.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/internaltest.cc

@@ -452 +452 @@
         Xapian::Query::Internal * qint;
 
         s = query->internal->serialise();
-        qint = Xapian::Query::Internal::unserialise(s);
+        map<string, Xapian::PostingSource *> m;
+        qint = Xapian::Query::Internal::unserialise(s, m);
 
         TEST(qint->serialise() == s);
         delete qint;

xapian-core/tests/api_db.cc

@@ -1894 +1894 @@
         : 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; }
 
@@ -1932 +1939 @@
     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));
 
     Xapian::MSet mset = enq.get_mset(0, 10);
@@ -1988 +1990 @@
         : 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;
     }
@@ -2085 +2094 @@
         : 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");
     }
@@ -2158 +2174 @@
 }
 
 // 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;
@@ -2196 +2209 @@
 // 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);
@@ -2211 +2222 @@
 
 // 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;

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.
+     *
+     *  This may return NULL to indicate that cloning is not supported.  In
+     *  this case, the PostingSource may only be used with a single-database
+     *  search.
+     *
+     *  The default implementation returns NULL.
+     */
+    virtual PostingSource * clone() const;
+
+    /** 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 +240 @@
     /// 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 +261 @@
      *  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 +278 @@
 
     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.
+        /** 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(Xapian::PostingSource * external_source);
 
         /** A query which matches all documents in the database. */
@@ -325 +332 @@
         /// External posting source.
         Xapian::PostingSource * external_source;
 
+        /// Flag, indicating whether the external source is owned by the query.
+        bool external_source_owned;
+
         /** swap the contents of this with another Xapian::Query::Internal,
          *  in a way which is guaranteed not to throw.  This is
          *  used with the assignment operator to make it exception
@@ -400 +410 @@
         Internal(op_t op_, Xapian::valueno valno, const std::string &value);
 
         /// Construct an external source query.
-        explicit Internal(Xapian::PostingSource * external_source_);
+        explicit Internal(Xapian::PostingSource * external_source_, bool owned);
 
         /** 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), true);
+}
+
 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 '!':
+                    subqs.push_back(readexternal());
+                    break;
                 case '(': {
                     subqs.push_back(readcompound());
                     break;
@@ -563 +606 @@
 }
 
 Xapian::Query::Internal *
-Xapian::Query::Internal::unserialise(const string &s)
+Xapian::Query::Internal::unserialise(const string &s,
+                        const map<string, Xapian::PostingSource *> & sources)
 {
     Assert(s.length() > 1);
-    QUnserial u(s);
+    QUnserial u(s, sources);
     Xapian::Query::Internal * qint = u.decode();
     AssertEq(s, qint->serialise());
     return qint;
 }
 #else
 Xapian::Query::Internal *
-Xapian::Query::Internal::unserialise(const string &)
+Xapian::Query::Internal::unserialise(const string &,
+                        const map<string, Xapian::PostingSource *> & sources)
 {
     throw Xapian::InternalError("query serialisation not compiled in");
 }
@@ -597 +642 @@
     std::swap(term_pos, other.term_pos);
     std::swap(wqf, other.wqf);
     std::swap(external_source, other.external_source);
+    std::swap(external_source_owned, other.external_source_owned);
 }
 
 Xapian::Query::Internal::Internal(const Xapian::Query::Internal &copyme)
@@ -608 +654 @@
           str_parameter(copyme.str_parameter),
           term_pos(copyme.term_pos),
           wqf(copyme.wqf),
-          external_source(copyme.external_source)
+          external_source(NULL),
+          external_source_owned(false)
 {
     for (subquery_list::const_iterator i = copyme.subqs.begin();
          i != copyme.subqs.end();
          ++i) {
         subqs.push_back(new Xapian::Query::Internal(**i));
     }
+    if (copyme.external_source) {
+        external_source = copyme.external_source->clone();
+        if (external_source == NULL) {
+            external_source = copyme.external_source;
+            external_source_owned = false;
+        } else {
+            external_source_owned = true;
+        }
+    }
 }
 
 //////////////////////////////////////////
@@ -627 +683 @@
           parameter(0),
           tname(tname_),
           term_pos(term_pos_),
-          wqf(wqf_)
+          wqf(wqf_),
+          external_source(NULL),
+          external_source_owned(false)
 {
     validate_query();
 }
@@ -638 +696 @@
           parameter(parameter_),
           tname(),
           term_pos(0),
-          wqf(0)
+          wqf(0),
+          external_source(NULL),
+          external_source_owned(false)
 {
     if (parameter != 0 && op != OP_PHRASE && op != OP_NEAR && op != OP_ELITE_SET)
         throw Xapian::InvalidArgumentError("parameter is only meaningful for OP_NEAR, OP_PHRASE, or OP_ELITE_SET");
@@ -649 +709 @@
         : op(op_),
           parameter(Xapian::termcount(valno)),
           tname(begin),
-          str_parameter(end)
+          str_parameter(end),
+          external_source(NULL),
+          external_source_owned(false)
 {
     if (op != OP_VALUE_RANGE)
         throw Xapian::InvalidArgumentError("This constructor is only meaningful for OP_VALUE_RANGE");
@@ -660 +722 @@
                                   const std::string &value)
         : op(op_),
           parameter(Xapian::termcount(valno)),
-          tname(value)
+          tname(value),
+          external_source(NULL),
+          external_source_owned(false)