You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by pg...@apache.org on 2023/10/05 17:09:33 UTC
[couchdb] 02/03: Revert "mango: prevent occasional duplication of paginated `text` results"
This is an automated email from the ASF dual-hosted git repository.
pgj pushed a commit to branch revert-4782-fix/mango/duplicated-text-results
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 05ad7058ba0786ed23ed752e8f549bd82d919b74
Author: PÁLI Gábor János <pa...@gmail.com>
AuthorDate: Thu Oct 5 19:09:24 2023 +0200
Revert "mango: prevent occasional duplication of paginated `text` results"
This reverts commit d6d4ad97921a50e0e8ef8a4dc872aefa93993469.
---
src/mango/src/mango_cursor_text.erl | 138 +++++++------------------------
src/mango/test/24-text-paginated-test.py | 108 ------------------------
src/mango/test/mango.py | 46 ++---------
3 files changed, 36 insertions(+), 256 deletions(-)
diff --git a/src/mango/src/mango_cursor_text.erl b/src/mango/src/mango_cursor_text.erl
index 09d878afd..f5899914a 100644
--- a/src/mango/src/mango_cursor_text.erl
+++ b/src/mango/src/mango_cursor_text.erl
@@ -37,8 +37,7 @@
user_fun,
user_acc,
fields,
- execution_stats,
- documents_seen
+ execution_stats
}).
create(Db, {Indexes, Trace}, Selector, Opts0) ->
@@ -109,8 +108,7 @@ execute(Cursor, UserFun, UserAcc) ->
user_fun = UserFun,
user_acc = UserAcc,
fields = Cursor#cursor.fields,
- execution_stats = mango_execution_stats:log_start(Stats),
- documents_seen = sets:new([{version, 2}])
+ execution_stats = mango_execution_stats:log_start(Stats)
},
try
case Query of
@@ -181,42 +179,28 @@ handle_hit(CAcc0, Sort, Doc) ->
#cacc{
limit = Limit,
skip = Skip,
- execution_stats = Stats,
- documents_seen = Seen
+ execution_stats = Stats
} = CAcc0,
+ CAcc1 = update_bookmark(CAcc0, Sort),
Stats1 = mango_execution_stats:incr_docs_examined(Stats),
couch_stats:increment_counter([mango, docs_examined]),
- CAcc1 = CAcc0#cacc{execution_stats = Stats1},
- case mango_selector:match(CAcc1#cacc.selector, Doc) of
- true ->
- DocId = mango_doc:get_field(Doc, <<"_id">>),
- case sets:is_element(DocId, Seen) of
- true ->
- CAcc1;
- false ->
- CAcc2 = update_bookmark(CAcc1, Sort),
- CAcc3 = CAcc2#cacc{
- documents_seen = sets:add_element(DocId, Seen)
- },
- if
- Skip > 0 ->
- CAcc3#cacc{skip = Skip - 1};
- Limit == 0 ->
- % We hit this case if the user specified
- % with a zero limit. Notice that in this
- % case we need to return the bookmark from
- % before this match.
- throw({stop, CAcc0});
- Limit == 1 ->
- CAcc4 = apply_user_fun(CAcc3, Doc),
- throw({stop, CAcc4});
- Limit > 1 ->
- CAcc4 = apply_user_fun(CAcc3, Doc),
- CAcc4#cacc{limit = Limit - 1}
- end
- end;
+ CAcc2 = CAcc1#cacc{execution_stats = Stats1},
+ case mango_selector:match(CAcc2#cacc.selector, Doc) of
+ true when Skip > 0 ->
+ CAcc2#cacc{skip = Skip - 1};
+ true when Limit == 0 ->
+ % We hit this case if the user spcified with a
+ % zero limit. Notice that in this case we need
+ % to return the bookmark from before this match
+ throw({stop, CAcc0});
+ true when Limit == 1 ->
+ NewCAcc = apply_user_fun(CAcc2, Doc),
+ throw({stop, NewCAcc});
+ true when Limit > 1 ->
+ NewCAcc = apply_user_fun(CAcc2, Doc),
+ NewCAcc#cacc{limit = Limit - 1};
false ->
- CAcc1
+ CAcc2
end.
apply_user_fun(CAcc, Doc) ->
@@ -493,7 +477,6 @@ execute_test_() ->
append_sort_type,
fun(RawField, selector) -> <<RawField/binary, "<type>">> end
),
- meck:expect(mango_doc, get_field, fun({doc, N}, <<"_id">>) -> N end),
meck:expect(mango_fields, extract, fun({doc, N}, fields) -> {final_doc, N} end),
meck:expect(
foo,
@@ -519,7 +502,6 @@ execute_test_() ->
?TDEF_FE(t_execute_more_results, 10),
?TDEF_FE(t_execute_unique_results, 10),
?TDEF_FE(t_execute_limit_cutoff, 10),
- ?TDEF_FE(t_execute_limit_cutoff_unique, 10),
?TDEF_FE(t_execute_limit_zero, 10),
?TDEF_FE(t_execute_limit_unique, 10),
?TDEF_FE(t_execute_skip, 10),
@@ -637,7 +619,6 @@ t_execute_more_results(_) ->
?assertEqual(3, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
t_execute_unique_results(_) ->
- UniqueHits = 3,
Options = [{partition, partition}, {sort, {[]}}, {bookmark, []}],
Cursor = #cursor{
db = db,
@@ -680,10 +661,10 @@ t_execute_unique_results(_) ->
),
meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
meck:expect(mango_selector, match, fun(selector, {doc, _}) -> true end),
- ?assertEqual({ok, {acc, 6}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
+ ?assertEqual({ok, {acc, 9}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
?assertEqual(6, meck:num_calls(couch_stats, increment_counter, 1)),
?assertEqual(6, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
- ?assertEqual(UniqueHits, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
+ ?assertEqual(6, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
t_execute_limit_cutoff(_) ->
Limit = 2,
@@ -727,57 +708,6 @@ t_execute_limit_cutoff(_) ->
?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
-t_execute_limit_cutoff_unique(_) ->
- Limit = 4,
- ActualHits = 3,
- AllHits = 6,
- Options = [{partition, partition}, {sort, {[]}}, {bookmark, []}],
- Cursor = #cursor{
- db = db,
- index = #idx{ddoc = <<"ddoc">>, name = <<"index">>},
- limit = Limit,
- skip = 0,
- fields = fields,
- selector = selector,
- opts = Options,
- execution_stats = stats
- },
- meck:expect(
- dreyfus_fabric_search,
- go,
- fun(db_name, <<"ddoc">>, <<"index">>, QueryArgs) ->
- #index_query_args{
- q = query,
- partition = partition,
- bookmark = B,
- sort = relevance,
- raw_bookmark = true
- } = QueryArgs,
- {Bookmark, Hits} =
- case B of
- nil ->
- Hit1 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 1}}]}},
- Hit2 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 2}}]}},
- Hit3 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 3}}]}},
- {[bookmark, 0], [Hit1, Hit2, Hit3]};
- [bookmark, 3] ->
- Hit1 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 1}}]}},
- Hit2 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 2}}]}},
- Hit3 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 3}}]}},
- {[bookmark, 4], [Hit3, Hit2, Hit1]};
- [bookmark, 4] ->
- {[bookmark, 5], []}
- end,
- {ok, Bookmark, undefined, Hits, undefined, undefined}
- end
- ),
- meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
- meck:expect(mango_selector, match, fun(selector, {doc, _}) -> true end),
- ?assertEqual({ok, {acc, 6}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
- ?assertEqual(AllHits, meck:num_calls(couch_stats, increment_counter, 1)),
- ?assertEqual(AllHits, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
- ?assertEqual(ActualHits, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
-
t_execute_limit_zero(_) ->
Limit = 0,
Skip = 0,
@@ -821,8 +751,6 @@ t_execute_limit_zero(_) ->
t_execute_limit_unique(_) ->
Limit = 5,
- AllHits = 6,
- UniqueHits = 3,
Options = [{partition, partition}, {sort, {[]}}, {bookmark, []}],
Cursor = #cursor{
db = db,
@@ -856,19 +784,17 @@ t_execute_limit_unique(_) ->
Hit1 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 1}}]}},
Hit2 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 2}}]}},
Hit3 = #sortable{item = #hit{fields = [{<<"_id">>, {id, 3}}]}},
- {[bookmark, 4], [Hit3, Hit2, Hit1]};
- [bookmark, 4] ->
- {[bookmark, 5], []}
+ {[bookmark, 4], [Hit3, Hit2, Hit1]}
end,
{ok, Bookmark, undefined, Hits, undefined, undefined}
end
),
meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
meck:expect(mango_selector, match, fun(selector, {doc, _}) -> true end),
- ?assertEqual({ok, {acc, 6}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
- ?assertEqual(AllHits, meck:num_calls(couch_stats, increment_counter, 1)),
- ?assertEqual(AllHits, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
- ?assertEqual(UniqueHits, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
+ ?assertEqual({ok, {acc, 8}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
+ ?assertEqual(Limit, meck:num_calls(couch_stats, increment_counter, 1)),
+ ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
+ ?assertEqual(Limit, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
t_execute_skip(_) ->
UniqueHits = 3,
@@ -919,14 +845,12 @@ t_execute_skip(_) ->
t_execute_skip_unique(_) ->
AllHits = 6,
- UniqueHits = 3,
- Skip = 2,
Options = [{partition, partition}, {sort, {[]}}, {bookmark, []}],
Cursor = #cursor{
db = db,
index = #idx{ddoc = <<"ddoc">>, name = <<"index">>},
limit = 10,
- skip = Skip,
+ skip = 2,
fields = fields,
selector = selector,
opts = Options,
@@ -963,12 +887,10 @@ t_execute_skip_unique(_) ->
),
meck:expect(mango_selector_text, convert, [selector], meck:val(query)),
meck:expect(mango_selector, match, fun(selector, {doc, _}) -> true end),
- ?assertEqual({ok, {acc, 4}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
+ ?assertEqual({ok, {acc, 7}}, execute(Cursor, fun foo:normal/2, {acc, 0})),
?assertEqual(AllHits, meck:num_calls(couch_stats, increment_counter, 1)),
?assertEqual(AllHits, meck:num_calls(mango_execution_stats, incr_docs_examined, 1)),
- ?assertEqual(
- UniqueHits - Skip, meck:num_calls(mango_execution_stats, incr_results_returned, 1)
- ).
+ ?assertEqual(4, meck:num_calls(mango_execution_stats, incr_results_returned, 1)).
t_execute_no_matches(_) ->
UniqueHits = 3,
diff --git a/src/mango/test/24-text-paginated-test.py b/src/mango/test/24-text-paginated-test.py
deleted file mode 100644
index 84b3a2f4b..000000000
--- a/src/mango/test/24-text-paginated-test.py
+++ /dev/null
@@ -1,108 +0,0 @@
-# Licensed under the Apache License, Version 2.0 (the "License"); you may not
-# use this file except in compliance with the License. You may obtain a copy of
-# the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations under
-# the License.
-
-
-import mango
-import unittest
-import time
-
-
-@unittest.skipUnless(mango.has_text_service(), "requires text service")
-class PaginatedResultsTest(mango.DbPerClass):
- # Great enough to make faster systems busy while running the
- # query.
- NUM_DOCS = 10000
- UPDATES = 25
-
- def setUp(self):
- self.db.recreate()
- self.db.create_text_index(
- analyzer="keyword",
- default_field={},
- selector={},
- fields=[
- {"name": "_id", "type": "string"},
- {"name": "name", "type": "string"},
- ],
- index_array_lengths=True,
- )
- docs = [
- {"_id": f"{doc_id:08X}", "name": mango.random_string(32)}
- for doc_id in range(self.NUM_DOCS)
- ]
- self.db.save_docs(docs)
-
- def test_query_with_lot_of_results(self):
- # 200 is the maximum for `text` searches.
- docs = self.db.find(selector={"_id": {"$lte": f"{1000:08X}"}}, limit=200)
- assert len(docs) == 200
-
- def do_query(self, delay, find_args):
- time.sleep(delay)
- return self.db.find(*find_args)
-
- def do_updates(self, pause, doc_id):
- for i in range(self.UPDATES):
- doc = self.db.open_doc(doc_id)
- updated_doc = {
- "_id": doc_id,
- "_rev": doc["_rev"],
- "update": i,
- "name": "foobar",
- }
- self.db.save_doc(updated_doc)
- time.sleep(pause)
-
- def test_no_duplicates_on_interleaved_updates(self):
- # Give ~500 ms head start for the updates before running the
- # query.
- query = mango.Concurrently(self.do_query, (0.5, ({"name": "foobar"},)))
- # Keep updating the target document in every 200 ms.
- updates = mango.Concurrently(self.do_updates, (0.2, f"{2:08X}"))
- docs = query.get_result()
- updates.join()
- assert len(docs) == 1
-
- def test_no_duplicates_on_interleaved_updates_heavy(self):
- query = mango.Concurrently(self.do_query, (0.5, ({"name": "foobar"},)))
- updates = [
- mango.Concurrently(self.do_updates, (0.05, f"{2:08X}")),
- mango.Concurrently(self.do_updates, (0.2, f"{3:08X}")),
- mango.Concurrently(self.do_updates, (0.3, f"{4:08X}")),
- mango.Concurrently(self.do_updates, (0.15, f"{5:08X}")),
- mango.Concurrently(self.do_updates, (0.1, f"{6:08X}")),
- ]
- docs = query.get_result()
- for ref in updates:
- ref.join()
- ids = list(map(lambda d: d["_id"], docs))
- assert sorted(ids) == [
- f"{2:08X}",
- f"{3:08X}",
- f"{4:08X}",
- f"{5:08X}",
- f"{6:08X}",
- ]
-
- def test_no_duplicates_on_interleaved_updates_with_limit_skip(self):
- query = mango.Concurrently(self.do_query, (0.5, ({"name": "foobar"}, 1, 3)))
- updates = [
- mango.Concurrently(self.do_updates, (0.05, f"{2:08X}")),
- mango.Concurrently(self.do_updates, (0.2, f"{3:08X}")),
- mango.Concurrently(self.do_updates, (0.3, f"{4:08X}")),
- mango.Concurrently(self.do_updates, (0.15, f"{5:08X}")),
- mango.Concurrently(self.do_updates, (0.1, f"{6:08X}")),
- ]
- docs = query.get_result()
- for ref in updates:
- ref.join()
- assert len(docs) == 1
diff --git a/src/mango/test/mango.py b/src/mango/test/mango.py
index b8a6cd9a5..2dff18f40 100644
--- a/src/mango/test/mango.py
+++ b/src/mango/test/mango.py
@@ -11,13 +11,10 @@
# the License.
import json
-import random
-import string
import time
import unittest
import uuid
import os
-import threading
import requests
@@ -31,18 +28,10 @@ COUCH_USER = os.environ.get("COUCH_USER")
COUCH_PASS = os.environ.get("COUCH_PASS")
-BULK_BATCH_SIZE = 500
-
-
def random_db_name():
return "mango_test_" + uuid.uuid4().hex
-def random_string(n_max):
- n = random.choice(range(n_max))
- return "".join(random.choice(string.ascii_letters) for _ in range(n))
-
-
def has_text_service():
features = requests.get(COUCH_HOST).json()["features"]
return "search" in features
@@ -58,27 +47,6 @@ def delay(n=5, t=0.5):
time.sleep(t)
-class Concurrently(object):
- def __init__(self, thread, thread_args, start=True):
- self.thread = threading.Thread(target=self.wrapper, args=(thread, thread_args))
- self.return_value = None
- if start:
- self.start()
-
- def wrapper(self, body, args):
- self.return_value = body(*args)
-
- def start(self):
- self.thread.start()
-
- def get_result(self):
- self.thread.join()
- return self.return_value
-
- def join(self):
- self.thread.join()
-
-
class Database(object):
def __init__(
self,
@@ -135,14 +103,12 @@ class Database(object):
r.raise_for_status()
def save_docs(self, docs, **kwargs):
- for offset in range(0, len(docs), BULK_BATCH_SIZE):
- chunk = docs[offset : (offset + BULK_BATCH_SIZE)]
- body = {"docs": chunk}
- r = self.sess.post(self.path("_bulk_docs"), json=body, params=kwargs)
- r.raise_for_status()
- for doc, result in zip(chunk, r.json()):
- doc["_id"] = result["id"]
- doc["_rev"] = result["rev"]
+ body = json.dumps({"docs": docs})
+ r = self.sess.post(self.path("_bulk_docs"), data=body, params=kwargs)
+ r.raise_for_status()
+ for doc, result in zip(docs, r.json()):
+ doc["_id"] = result["id"]
+ doc["_rev"] = result["rev"]
def open_doc(self, docid):
r = self.sess.get(self.path(docid))