Merge lp:~thisfred/u1db/its-full-of-conflicts into lp:u1db

[Eric Casteleijn]

- refactored the way sql queries are built in the python backend, and brought it more into step with the C implementation.
- made get_all_docs, get_from_index and get_range_from_index conflict aware

- refactored the way sql queries are built in the python backend, and brought it more into step with the C implementation.
- made get_all_docs, get_from_index and get_range_from_index conflict aware

[lp:~thisfred/u1db/its-full-of-conflicts]

380. By Eric Casteleijn on 2012-08-09

clarified?

381. By Eric Casteleijn on 2012-08-10

no need to cast int

[John O'Brien (jdobrien)]

the queries look sane to me :)

review: Approve

=== modified file 'doc/sqlite_schema.txt'
--- doc/sqlite_schema.txt	2012-07-16 16:20:53 +0000
+++ doc/sqlite_schema.txt	2012-08-10 20:11:19 +0000
@@ -397,8 +397,8 @@
 database. We timed a loop that extracted every document from the database,
 one-at-a-time by calling ``get_doc()``. This also has the overhead of checking
 if a given document has conflicts, etc. All currently implementations also
-support ``_get_doc()`` which extracts the content without checking for
-conflicts. Further, `Partial Expanded Fields`_ internally has a
+support ``_get_doc()`` which extracts the content without (by default) checking
+for conflicts. Further, `Partial Expanded Fields`_ internally has a
 ``_iter_all_docs`` method that it uses when it wants to update the index data.
 This simply streams out the document content, as quickly as SQLite can produce
 it.

=== modified file 'src/u1db.c'
--- src/u1db.c	2012-07-31 19:43:46 +0000
+++ src/u1db.c	2012-08-10 20:11:19 +0000
@@ -1208,6 +1208,7 @@
                   void *context, u1db_doc_callback cb)
 {
     int status;
+    int conflicts;
     sqlite3_stmt *statement;
 
     if (db == NULL || cb == NULL) {
@@ -1217,7 +1218,10 @@
     if (status != U1DB_OK)
         return status;
     status = sqlite3_prepare_v2(db->sql_handle,
-        "SELECT doc_id, doc_rev, content FROM document", -1, &statement, NULL);
+        "SELECT document.doc_id, document.doc_rev, document.content, "
+        "count(conflicts.doc_rev) FROM document LEFT OUTER JOIN conflicts ON "
+        "conflicts.doc_id = document.doc_id GROUP BY document.doc_id, "
+        "document.doc_rev, document.content", -1, &statement, NULL);
     if (status != SQLITE_OK) { goto finish; }
     status = sqlite3_step(statement);
     while (status == SQLITE_ROW) {
@@ -1228,9 +1232,10 @@
         doc_id = (char *)sqlite3_column_text(statement, 0);
         revision = (char *)sqlite3_column_text(statement, 1);
         content = (char *)sqlite3_column_text(statement, 2);
+        conflicts = sqlite3_column_int(statement, 3);
         if (content != NULL || include_deleted) {
             status = u1db__allocate_document(
-                doc_id, revision, content, 0, &doc);
+                doc_id, revision, content, conflicts > 0, &doc);
             if (status != U1DB_OK)
                 goto finish;
             cb(context, doc);

=== modified file 'src/u1db_query.c'
--- src/u1db_query.c	2012-07-27 19:06:03 +0000
+++ src/u1db_query.c	2012-08-10 20:11:19 +0000
@@ -747,7 +747,7 @@
         // We use u1db_get_docs so we can pass check_for_conflicts=0, which is
         // currently expected by the test suite.
         status = u1db_get_docs(
-            db, 1, (const char**)&doc_id, 0, 0, context, cb);
+            db, 1, (const char**)&doc_id, 1, 0, context, cb);
         if (status != U1DB_OK) { goto finish; }
         status = sqlite3_step(statement);
     }
@@ -848,7 +848,7 @@
         // We use u1db_get_docs so we can pass check_for_conflicts=0, which is
         // currently expected by the test suite.
         status = u1db_get_docs(
-            db, 1, (const char**)&doc_id, 0, 0, context, cb);
+            db, 1, (const char**)&doc_id, 1, 0, context, cb);
         if (status != U1DB_OK) { goto finish; }
         status = sqlite3_step(statement);
     }

=== modified file 'u1db/backends/__init__.py'
--- u1db/backends/__init__.py	2012-07-26 19:14:26 +0000
+++ u1db/backends/__init__.py	2012-08-10 20:11:19 +0000
@@ -73,11 +73,10 @@
         """
         raise NotImplementedError(self._get_generation_info)
 
-    def _get_doc(self, doc_id):
+    def _get_doc(self, doc_id, check_for_conflicts=False):
         """Extract the document from storage.
 
-        This can return None if the document doesn't exist, it should not check
-        if there are any conflicts, etc.
+        This can return None if the document doesn't exist.
         """
         raise NotImplementedError(self._get_doc)
 
@@ -110,11 +109,10 @@
     def get_docs(self, doc_ids, check_for_conflicts=True,
                  include_deleted=False):
         for doc_id in doc_ids:
-            doc = self._get_doc(doc_id)
+            doc = self._get_doc(
+                doc_id, check_for_conflicts=check_for_conflicts)
             if doc.is_tombstone() and not include_deleted:
                 continue
-            if check_for_conflicts:
-                doc.has_conflicts = self._has_conflicts(doc_id)
             yield doc
 
     def _get_trans_id_for_gen(self, generation):

=== modified file 'u1db/backends/inmemory.py'
--- u1db/backends/inmemory.py	2012-07-13 19:35:51 +0000
+++ u1db/backends/inmemory.py	2012-08-10 20:11:19 +0000
@@ -100,9 +100,9 @@
             raise errors.InvalidDocId()
         self._check_doc_id(doc.doc_id)
         self._check_doc_size(doc)
-        if self._has_conflicts(doc.doc_id):
+        old_doc = self._get_doc(doc.doc_id, check_for_conflicts=True)
+        if old_doc and old_doc.has_conflicts:
             raise errors.ConflictedDoc()
-        old_doc = self._get_doc(doc.doc_id)
         if old_doc and doc.rev is None and old_doc.is_tombstone():
             new_rev = self._allocate_doc_rev(old_doc.rev)
         else:
@@ -127,23 +127,25 @@
         self._docs[doc.doc_id] = (doc.rev, doc.get_json())
         self._transaction_log.append((doc.doc_id, trans_id))
 
-    def _get_doc(self, doc_id):
+    def _get_doc(self, doc_id, check_for_conflicts=False):
         try:
             doc_rev, content = self._docs[doc_id]
         except KeyError:
             return None
-        return self._factory(doc_id, doc_rev, content)
+        doc = self._factory(doc_id, doc_rev, content)
+        if check_for_conflicts:
+            doc.has_conflicts = (doc.doc_id in self._conflicts)
+        return doc
 
     def _has_conflicts(self, doc_id):
         return doc_id in self._conflicts
 
     def get_doc(self, doc_id, include_deleted=False):
-        doc = self._get_doc(doc_id)
+        doc = self._get_doc(doc_id, check_for_conflicts=True)
         if doc is None:
             return None
         if doc.is_tombstone() and not include_deleted:
             return None
-        doc.has_conflicts = (doc.doc_id in self._conflicts)
         return doc
 
     def get_all_docs(self, include_deleted=False):
@@ -153,7 +155,9 @@
         for doc_id, (doc_rev, content) in self._docs.items():
             if content is None and not include_deleted:
                 continue
-            results.append(self._factory(doc_id, doc_rev, content))
+            doc = self._factory(doc_id, doc_rev, content)
+            doc.has_conflicts = self._has_conflicts(doc_id)
+            results.append(doc)
         return (generation, results)
 
     def get_doc_conflicts(self, doc_id):
@@ -249,8 +253,7 @@
         doc_ids = index.lookup(key_values)
         result = []
         for doc_id in doc_ids:
-            doc_rev, doc = self._docs[doc_id]
-            result.append(self._factory(doc_id, doc_rev, doc))
+            result.append(self._get_doc(doc_id, check_for_conflicts=True))
         return result
 
     def get_range_from_index(self, index_name, start_value=None,
@@ -267,8 +270,7 @@
         doc_ids = index.lookup_range(start_value, end_value)
         result = []
         for doc_id in doc_ids:
-            doc_rev, doc = self._docs[doc_id]
-            result.append(self._factory(doc_id, doc_rev, doc))
+            result.append(self._get_doc(doc_id, check_for_conflicts=True))
         return result
 
     def get_index_keys(self, index_name):

=== modified file 'u1db/backends/sqlite_backend.py'
--- u1db/backends/sqlite_backend.py	2012-08-01 16:41:14 +0000
+++ u1db/backends/sqlite_backend.py	2012-08-10 20:11:19 +0000
@@ -292,16 +292,27 @@
                   " ORDER BY generation")
         return c.fetchall()
 
-    def _get_doc(self, doc_id):
+    def _get_doc(self, doc_id, check_for_conflicts=False):
         """Get just the document content, without fancy handling."""
         c = self._db_handle.cursor()
-        c.execute("SELECT doc_rev, content FROM document WHERE doc_id = ?",
-                  (doc_id,))
+        if check_for_conflicts:
+            c.execute(
+                "SELECT document.doc_rev, document.content, "
+                "count(conflicts.doc_rev) FROM document LEFT OUTER JOIN "
+                "conflicts ON conflicts.doc_id = document.doc_id WHERE "
+                "document.doc_id = ? GROUP BY document.doc_id, "
+                "document.doc_rev, document.content;", (doc_id,))
+        else:
+            c.execute(
+                "SELECT doc_rev, content, 0 FROM document WHERE doc_id = ?",
+                (doc_id,))
         val = c.fetchone()
         if val is None:
             return None
-        doc_rev, content = val
-        return self._factory(doc_id, doc_rev, content)
+        doc_rev, content, conflicts = val
+        doc = self._factory(doc_id, doc_rev, content)
+        doc.has_conflicts = conflicts > 0
+        return doc
 
     def _has_conflicts(self, doc_id):
         c = self._db_handle.cursor()
@@ -314,13 +325,11 @@
             return True
 
     def get_doc(self, doc_id, include_deleted=False):
-        doc = self._get_doc(doc_id)
+        doc = self._get_doc(doc_id, check_for_conflicts=True)
         if doc is None:
             return None
         if doc.is_tombstone() and not include_deleted:
             return None
-        # TODO: A doc which appears deleted could still have conflicts...
-        doc.has_conflicts = self._has_conflicts(doc.doc_id)
         return doc
 
     def get_all_docs(self, include_deleted=False):
@@ -328,12 +337,18 @@
         generation = self._get_generation()
         results = []
         c = self._db_handle.cursor()
-        c.execute("SELECT doc_id, doc_rev, content FROM document;")
+        c.execute(
+            "SELECT document.doc_id, document.doc_rev, document.content, "
+            "count(conflicts.doc_rev) FROM document LEFT OUTER JOIN conflicts "
+            "ON conflicts.doc_id = document.doc_id GROUP BY document.doc_id, "
+            "document.doc_rev, document.content;")
         rows = c.fetchall()
-        for doc_id, doc_rev, content in rows:
+        for doc_id, doc_rev, content, conflicts in rows:
             if content is None and not include_deleted:
                 continue
-            results.append(self._factory(doc_id, doc_rev, content))
+            doc = self._factory(doc_id, doc_rev, content)
+            doc.has_conflicts = conflicts > 0
+            results.append(doc)
         return (generation, results)
 
     def put_doc(self, doc):
@@ -342,9 +357,9 @@
         self._check_doc_id(doc.doc_id)
         self._check_doc_size(doc)
         with self._db_handle:
-            if self._has_conflicts(doc.doc_id):
+            old_doc = self._get_doc(doc.doc_id, check_for_conflicts=True)
+            if old_doc and old_doc.has_conflicts:
                 raise errors.ConflictedDoc()
-            old_doc = self._get_doc(doc.doc_id)
             if old_doc and doc.rev is None and old_doc.is_tombstone():
                 new_rev = self._allocate_doc_rev(old_doc.rev)
             else:
@@ -429,14 +444,14 @@
 
     def delete_doc(self, doc):
         with self._db_handle:
-            old_doc = self._get_doc(doc.doc_id)
+            old_doc = self._get_doc(doc.doc_id, check_for_conflicts=True)
             if old_doc is None:
                 raise errors.DocumentDoesNotExist
             if old_doc.rev != doc.rev:
                 raise errors.RevisionConflict()
             if old_doc.is_tombstone():
                 raise errors.DocumentAlreadyDeleted
-            if self._has_conflicts(doc.doc_id):
+            if old_doc.has_conflicts:
                 raise errors.ConflictedDoc()
             new_rev = self._allocate_doc_rev(doc.rev)
             doc.rev = new_rev
@@ -588,22 +603,12 @@
         return fields
 
     @staticmethod
-    def _transform_glob(value, escape_char='.'):
-        """Transform the given glob value into a valid LIKE statement."""
-        to_escape = [escape_char, '%', '_']
-        for esc in to_escape:
-            value = value.replace(esc, escape_char + esc)
-        assert value[-1] == '*'
-        return value[:-1] + '%'
-
-    @staticmethod
     def _strip_glob(value):
         """Remove the trailing * from a value."""
         assert value[-1] == '*'
         return value[:-1]
 
-    def get_from_index(self, index_name, *key_values):
-        definition = self._get_index_definition(index_name)
+    def _format_query(self, definition, key_values):
         # First, build the definition. We join the document_fields table
         # against itself, as many times as the 'width' of our definition.
         # We then do a query for each key_value, one-at-a-time.
@@ -619,9 +624,8 @@
                        + (" AND d%d.value = ?" % (i,))
                        for i in range(len(definition))]
         like_where = [novalue_where[i]
-                      + (" AND d%d.value LIKE ? ESCAPE '.'" % (i,))
+                      + (" AND d%d.value GLOB ?" % (i,))
                       for i in range(len(definition))]
-        c = self._db_handle.cursor()
         is_wildcard = False
         # Merge the lists together, so that:
         # [field1, field2, field3], [val1, val2, val3]
@@ -629,8 +633,6 @@
         # (field1, val1, field2, val2, field3, val3)
         args = []
         where = []
-        if len(key_values) != len(definition):
-            raise errors.InvalidValueForIndex()
         for idx, (field, value) in enumerate(zip(definition, key_values)):
             args.append(field)
             if value.endswith('*'):
@@ -643,7 +645,7 @@
                         # another wildcard
                         raise errors.InvalidGlobbing
                     where.append(like_where[idx])
-                    args.append(self._transform_glob(value))
+                    args.append(value)
                 is_wildcard = True
             else:
                 if is_wildcard:
@@ -651,23 +653,33 @@
                 where.append(exact_where[idx])
                 args.append(value)
         statement = (
-            "SELECT d.doc_id, d.doc_rev, d.content FROM document d, %s "
-            "WHERE %s ORDER BY %s;" % (
-                ', '.join(tables), ' AND '.join(where),
-                ', '.join(
-                    ['d%d.value' % i for i in range(len(definition))])))
+            "SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM "
+            "document d, %s LEFT OUTER JOIN conflicts c ON c.doc_id = "
+            "d.doc_id WHERE %s GROUP BY d.doc_id, d.doc_rev, d.content ORDER "
+            "BY %s;" % (', '.join(tables), ' AND '.join(where), ', '.join(
+                ['d%d.value' % i for i in range(len(definition))])))
+        return statement, args
+
+    def get_from_index(self, index_name, *key_values):
+        definition = self._get_index_definition(index_name)
+        if len(key_values) != len(definition):
+            raise errors.InvalidValueForIndex()
+        statement, args = self._format_query(definition, key_values)
+        c = self._db_handle.cursor()
         try:
             c.execute(statement, tuple(args))
         except dbapi2.OperationalError, e:
             raise dbapi2.OperationalError(str(e) +
                 '\nstatement: %s\nargs: %s\n' % (statement, args))
         res = c.fetchall()
-        return [self._factory(r[0], r[1], r[2]) for r in res]
+        results = []
+        for row in res:
+            doc = self._factory(row[0], row[1], row[2])
+            doc.has_conflicts = row[3] > 0
+            results.append(doc)
+        return results
 
-    def get_range_from_index(self, index_name, start_value=None,
-                             end_value=None):
-        """Return all documents with key values in the specified range."""
-        definition = self._get_index_definition(index_name)
+    def _format_range_query(self, definition, start_value, end_value):
         tables = ["document_fields d%d" % i for i in range(len(definition))]
         novalue_where = [
             "d.doc_id = d%d.doc_id AND d%d.field_name = ?" % (i, i) for i in
@@ -677,9 +689,8 @@
             range(len(definition))]
         like_where = [
             novalue_where[i] + (
-                " AND (d%d.value < ? OR d%d.value LIKE ? ESCAPE '.')" %
-                (i, i))
-            for i in range(len(definition))]
+                " AND (d%d.value < ? OR d%d.value GLOB ?)" % (i, i)) for i in
+            range(len(definition))]
         range_where_lower = [
             novalue_where[i] + (" AND d%d.value >= ?" % (i,)) for i in
             range(len(definition))]
@@ -732,7 +743,7 @@
                             raise errors.InvalidGlobbing
                         where.append(like_where[idx])
                         args.append(self._strip_glob(value))
-                        args.append(self._transform_glob(value))
+                        args.append(value)
                     is_wildcard = True
                 else:
                     if is_wildcard:
@@ -740,12 +751,19 @@
                     where.append(range_where_upper[idx])
                     args.append(value)
         statement = (
-            "SELECT d.doc_id, d.doc_rev, d.content FROM document d, %s "
-            "WHERE %s ORDER BY %s;" % (
-                ', '.join(tables),
-                ' AND '.join(where),
-                ', '.join(
-                    ['d%d.value' % i for i in range(len(definition))])))
+            "SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM "
+            "document d, %s LEFT OUTER JOIN conflicts c ON c.doc_id = "
+            "d.doc_id WHERE %s GROUP BY d.doc_id, d.doc_rev, d.content ORDER "
+            "BY %s;" % (', '.join(tables), ' AND '.join(where), ', '.join(
+                ['d%d.value' % i for i in range(len(definition))])))
+        return statement, args
+
+    def get_range_from_index(self, index_name, start_value=None,
+                             end_value=None):
+        """Return all documents with key values in the specified range."""
+        definition = self._get_index_definition(index_name)
+        statement, args = self._format_range_query(
+            definition, start_value, end_value)
         c = self._db_handle.cursor()
         try:
             c.execute(statement, tuple(args))
@@ -753,7 +771,12 @@
             raise dbapi2.OperationalError(str(e) +
                 '\nstatement: %s\nargs: %s\n' % (statement, args))
         res = c.fetchall()
-        return [self._factory(r[0], r[1], r[2]) for r in res]
+        results = []
+        for row in res:
+            doc = self._factory(row[0], row[1], row[2])
+            doc.has_conflicts = row[3] > 0
+            results.append(doc)
+        return results
 
     def get_index_keys(self, index_name):
         c = self._db_handle.cursor()

=== modified file 'u1db/tests/test_backends.py'
--- u1db/tests/test_backends.py	2012-08-01 17:07:36 +0000
+++ u1db/tests/test_backends.py	2012-08-10 20:11:19 +0000
@@ -667,6 +667,15 @@
         self.assertEqual([alt_doc, doc],
                          self.db.get_doc_conflicts(doc.doc_id))
 
+    def test_get_all_docs_sees_conflicts(self):
+        doc = self.db.create_doc_from_json(simple_doc)
+        alt_doc = self.make_document(doc.doc_id, 'alternate:1', nested_doc)
+        self.db._put_doc_if_newer(
+            alt_doc, save_conflict=True, replica_uid='r', replica_gen=1,
+            replica_trans_id='foo')
+        _, docs = self.db.get_all_docs()
+        self.assertTrue(docs[0].has_conflicts)
+
     def test_get_doc_conflicts_unconflicted(self):
         doc = self.db.create_doc_from_json(simple_doc)
         self.assertEqual([], self.db.get_doc_conflicts(doc.doc_id))
@@ -1124,6 +1133,18 @@
              ('value', 'value2-3')],
             sorted(self.db.get_index_keys('test-idx')))
 
+    def test_get_from_index_sees_conflicts(self):
+        doc = self.db.create_doc_from_json(simple_doc)
+        self.db.create_index('test-idx', 'key', 'key2')
+        alt_doc = self.make_document(
+            doc.doc_id, 'alternate:1',
+            '{"key": "value", "key2": ["value2-1", "value2-2", "value2-3"]}')
+        self.db._put_doc_if_newer(
+            alt_doc, save_conflict=True, replica_uid='r', replica_gen=1,
+            replica_trans_id='foo')
+        docs = self.db.get_from_index('test-idx', 'value', 'value2-1')
+        self.assertTrue(docs[0].has_conflicts)
+
     def test_get_index_keys_multi_list_list(self):
         self.db.create_doc_from_json(
             '{"key": "value1-1 value1-2 value1-3", '
@@ -1171,6 +1192,17 @@
             [doc2, doc1, doc3],
             self.db.get_range_from_index('test-idx', 'value2'))
 
+    def test_get_range_from_index_sees_conflicts(self):
+        doc = self.db.create_doc_from_json(simple_doc)
+        self.db.create_index('test-idx', 'key')
+        alt_doc = self.make_document(
+            doc.doc_id, 'alternate:1', '{"key": "valuedepalue"}')
+        self.db._put_doc_if_newer(
+            alt_doc, save_conflict=True, replica_uid='r', replica_gen=1,
+            replica_trans_id='foo')
+        docs = self.db.get_range_from_index('test-idx', 'a')
+        self.assertTrue(docs[0].has_conflicts)
+
     def test_get_range_from_index_end(self):
         self.db.create_doc_from_json('{"key": "value3"}')
         doc2 = self.db.create_doc_from_json('{"key": "value2"}')

=== modified file 'u1db/tests/test_sqlite_backend.py'
--- u1db/tests/test_sqlite_backend.py	2012-07-31 14:36:19 +0000
+++ u1db/tests/test_sqlite_backend.py	2012-08-10 20:11:19 +0000
@@ -360,18 +360,6 @@
         self.assertRaises(errors.DatabaseDoesNotExist,
                           sqlite_backend.SQLiteDatabase.delete_database, path)
 
-    def assertTransform(self, sql_value, value):
-        transformed = sqlite_backend.SQLiteDatabase._transform_glob(value)
-        self.assertEqual(sql_value, transformed)
-
-    def test_glob_escaping(self):
-        # SQL allows us to define any escape char we want, for now I'm just
-        # using '.'
-        self.assertTransform('val%', 'val*')
-        self.assertTransform('v.%al%', 'v%al*')
-        self.assertTransform('v._al%', 'v_al*')
-        self.assertTransform('v..al%', 'v.al*')
-
     def test__get_indexed_fields(self):
         self.db.create_index('idx1', 'a', 'b')
         self.assertEqual(set(['a', 'b']), self.db._get_indexed_fields())
@@ -395,3 +383,111 @@
         c.execute("SELECT doc_id, field_name, value FROM document_fields"
                   " ORDER BY doc_id, field_name, value")
         self.assertEqual([(doc1.doc_id, 'key1', 'val1')], c.fetchall())
+
+    def assertFormatQueryEquals(self, exp_statement, exp_args, definition,
+                                values):
+        statement, args = self.db._format_query(definition, values)
+        self.assertEqual(exp_statement, statement)
+        self.assertEqual(exp_args, args)
+
+    def test__format_query(self):
+        self.assertFormatQueryEquals(
+            "SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM "
+            "document d, document_fields d0 LEFT OUTER JOIN conflicts c ON "
+            "c.doc_id = d.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name "
+            "= ? AND d0.value = ? GROUP BY d.doc_id, d.doc_rev, d.content "
+            "ORDER BY d0.value;", ["key1", "a"],
+            ["key1"], ["a"])
+
+    def test__format_query2(self):
+        self.assertFormatQueryEquals(
+            'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM '
+            'document d, document_fields d0, document_fields d1, '
+            'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = '
+            'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value = ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value = ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND '
+            'd2.value = ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY '
+            'd0.value, d1.value, d2.value;',
+            ["key1", "a", "key2", "b", "key3", "c"],
+            ["key1", "key2", "key3"], ["a", "b", "c"])
+
+    def test__format_query_wildcard(self):
+        self.assertFormatQueryEquals(
+            'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM '
+            'document d, document_fields d0, document_fields d1, '
+            'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = '
+            'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value = ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value GLOB ? AND d.doc_id = d2.doc_id AND d2.field_name = ? '
+            'AND d2.value NOT NULL GROUP BY d.doc_id, d.doc_rev, d.content '
+            'ORDER BY d0.value, d1.value, d2.value;',
+            ["key1", "a", "key2", "b*", "key3"], ["key1", "key2", "key3"],
+            ["a", "b*", "*"])
+
+    def assertFormatRangeQueryEquals(self, exp_statement, exp_args, definition,
+                                     start_value, end_value):
+        statement, args = self.db._format_range_query(
+            definition, start_value, end_value)
+        self.assertEqual(exp_statement, statement)
+        self.assertEqual(exp_args, args)
+
+    def test__format_range_query(self):
+        self.assertFormatRangeQueryEquals(
+            'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM '
+            'document d, document_fields d0, document_fields d1, '
+            'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = '
+            'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value >= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value >= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND '
+            'd2.value >= ? AND d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value <= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value <= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND '
+            'd2.value <= ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY '
+            'd0.value, d1.value, d2.value;',
+            ['key1', 'a', 'key2', 'b', 'key3', 'c', 'key1', 'p', 'key2', 'q',
+             'key3', 'r'],
+            ["key1", "key2", "key3"], ["a", "b", "c"], ["p", "q", "r"])
+
+    def test__format_range_query_no_start(self):
+        self.assertFormatRangeQueryEquals(
+            'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM '
+            'document d, document_fields d0, document_fields d1, '
+            'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = '
+            'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value <= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value <= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND '
+            'd2.value <= ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY '
+            'd0.value, d1.value, d2.value;',
+            ['key1', 'a', 'key2', 'b', 'key3', 'c'],
+            ["key1", "key2", "key3"], None, ["a", "b", "c"])
+
+    def test__format_range_query_no_end(self):
+        self.assertFormatRangeQueryEquals(
+            'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM '
+            'document d, document_fields d0, document_fields d1, '
+            'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = '
+            'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value >= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value >= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND '
+            'd2.value >= ? GROUP BY d.doc_id, d.doc_rev, d.content ORDER BY '
+            'd0.value, d1.value, d2.value;',
+            ['key1', 'a', 'key2', 'b', 'key3', 'c'],
+            ["key1", "key2", "key3"], ["a", "b", "c"], None)
+
+    def test__format_range_query_wildcard(self):
+        self.assertFormatRangeQueryEquals(
+            'SELECT d.doc_id, d.doc_rev, d.content, count(c.doc_rev) FROM '
+            'document d, document_fields d0, document_fields d1, '
+            'document_fields d2 LEFT OUTER JOIN conflicts c ON c.doc_id = '
+            'd.doc_id WHERE d.doc_id = d0.doc_id AND d0.field_name = ? AND '
+            'd0.value >= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? AND '
+            'd1.value >= ? AND d.doc_id = d2.doc_id AND d2.field_name = ? AND '
+            'd2.value NOT NULL AND d.doc_id = d0.doc_id AND d0.field_name = ? '
+            'AND d0.value <= ? AND d.doc_id = d1.doc_id AND d1.field_name = ? '
+            'AND (d1.value < ? OR d1.value GLOB ?) AND d.doc_id = d2.doc_id '
+            'AND d2.field_name = ? AND d2.value NOT NULL GROUP BY d.doc_id, '
+            'd.doc_rev, d.content ORDER BY d0.value, d1.value, d2.value;',
+            ['key1', 'a', 'key2', 'b', 'key3', 'key1', 'p', 'key2', 'q', 'q*',
+             'key3'],
+            ["key1", "key2", "key3"], ["a", "b*", "*"], ["p", "q*", "*"])

=== modified file 'u1db/tests/test_sync.py'
--- u1db/tests/test_sync.py	2012-07-19 19:50:58 +0000
+++ u1db/tests/test_sync.py	2012-08-10 20:11:19 +0000
@@ -816,7 +816,10 @@
         self.assertTransactionLog([doc_id, doc_id], self.db1)
         self.assertGetDoc(self.db1, doc_id, doc2_rev, new_doc, True)
         self.assertGetDoc(self.db2, doc_id, doc2_rev, new_doc, False)
-        self.assertEqual([doc2], self.db1.get_from_index('test-idx', 'altval'))
+        from_idx = self.db1.get_from_index('test-idx', 'altval')[0]
+        self.assertEqual(doc2.doc_id, from_idx.doc_id)
+        self.assertEqual(doc2.rev, from_idx.rev)
+        self.assertTrue(from_idx.has_conflicts)
         self.assertEqual([], self.db1.get_from_index('test-idx', 'value'))
 
     def test_sync_sees_remote_delete_conflicted(self):
@@ -866,7 +869,10 @@
         self.sync(self.db1, self.db2, trace_hook=after_whatschanged)
         self.assertEqual([True], triggered)
         self.assertGetDoc(self.db1, doc_id, doc2_rev2, content2, True)
-        self.assertEqual([doc], self.db1.get_from_index('test-idx', 'altval'))
+        from_idx = self.db1.get_from_index('test-idx', 'altval')[0]
+        self.assertEqual(doc.doc_id, from_idx.doc_id)
+        self.assertEqual(doc.rev, from_idx.rev)
+        self.assertTrue(from_idx.has_conflicts)
         self.assertEqual([], self.db1.get_from_index('test-idx', 'value'))
         self.assertEqual([], self.db1.get_from_index('test-idx', 'localval'))