Repository: frappe/frappe
Pull request: perf(GlobalSearch): bulk insert queued items
Fixes #14660

Global search queue items are popped and inserted into the `__global_search`
table. This used to happen individually.

This PR bulk inserts them in chunks of 10,000.

Problem statement:
sync_global_search should do bulk insert
I'm in the process of importing the data from our current system into ERPNext, which includes 60,000+ items. I am having to do these 5000 lines at a time because of the timeout limit, but I have also noticed that `frappe.utils.global_search.sync_global_search` seems to be called for every single new record. This is a huge drain on resources and as there are only two workers on the background jobs I have to wait for these to be processed before doing anything else.

I think the system should be smart enough to know that if an import is running, then sync_global_search should not run until the import has finished - and then it should only run once.

Hints from issue discussion:
While this is improved and the original problem you have mentioned is gone; it still does one insert at a time 😄 Maybe this can be further optimized to do bulk insert!

https://github.com/frappe/frappe/blob/3a34bfc52035b2ff23091d827d8f41a8c7f49f30/frappe/utils/global_search.py#L355-L357

Commit message: perf(GlobalSearch): bulk insert queued items

Fixes #14660

Patch to review:
diff --git a/frappe/utils/global_search.py b/frappe/utils/global_search.py
index d310e1e8d56f..9626d4aa1581 100644
--- a/frappe/utils/global_search.py
+++ b/frappe/utils/global_search.py
@@ -373,11 +373,42 @@ def sync_global_search():
 	:param flags:
 	:return:
 	"""
-	while frappe.cache.llen("global_search_queue") > 0:
-		# rpop to follow FIFO
-		# Last one should override all previous contents of same document
-		value = json.loads(frappe.cache.rpop("global_search_queue").decode("utf-8"))
-		sync_value(value)
+	from itertools import islice
+
+	def get_search_queue_items_generator():
+		while frappe.cache.llen("global_search_queue") > 0:
+			item_json = frappe.cache.rpop("global_search_queue").decode("utf-8")
+			item_values = json.loads(item_json).values()
+			yield tuple(item_values)
+
+	search_queue_items = get_search_queue_items_generator()
+
+	while values := tuple(islice(search_queue_items, 10_000)):
+		sync_values(values)
+
+
+def sync_values(values):
+	from pypika.terms import Values
+
+	GlobalSearch = frappe.qb.Table("__global_search")
+	conflict_fields = ["content", "published", "title", "route"]
+
+	query = (
+		frappe.qb.into(GlobalSearch).columns(["doctype", "name"] + conflict_fields).insert(*values)
+	)
+
+	if frappe.db.db_type == "postgres":
+		query.on_conflict(GlobalSearch.doctype, GlobalSearch.name)
+
+	for field in conflict_fields:
+		if frappe.db.db_type == "mariadb":
+			query.on_duplicate_key_update(GlobalSearch[field], Values(field))
+		elif frappe.db.db_type == "postgres":
+			query.do_update(GlobalSearch[field], Values(field))
+		else:
+			raise NotImplementedError
+
+	query.run()
 
 
 def sync_value_in_queue(value):


Code review comments:
diff_hunk: @@ -373,11 +373,42 @@ def sync_global_search():
 	:param flags:
 	:return:
 	"""
-	while frappe.cache.llen("global_search_queue") > 0:
-		# rpop to follow FIFO
-		# Last one should override all previous contents of same document
-		value = json.loads(frappe.cache.rpop("global_search_queue").decode("utf-8"))
-		sync_value(value)
+	from itertools import islice
+
+	def get_search_queue_items_generator():
+		while frappe.cache.llen("global_search_queue") > 0:
+			item_json = frappe.cache.rpop("global_search_queue").decode("utf-8")
+			item_values = json.loads(item_json).values()
+			yield tuple(item_values)
+
+	search_queue_items = get_search_queue_items_generator()
+
+	while values := tuple(islice(search_queue_items, 10_000)):
path: frappe/utils/global_search.py
text: @user1:
There's create batch utils somewhere in framework.

Also can we somehow pop from redis in batches too? 🤔 

(Need to ensure that we don't miss anything though)

@user1:
Another question: what happens when same document gets modified twice and is sent twice via same query. DBs might not like it 😄 

@author:
`RPOP KEY LEN` can performed, will have to alter the wrapper though, should I do that?

@author:
Okay, nvm the `LEN` argument was added in 6.20 we (CI, FC) run an older version of redis which doesn't have that.

@author:
Also `utils.create_batch` won't work for this purpose cause while it asks for an `Iterable`, before yielding it slices the `Iterable`.

`Iterable` does not imply sliceable.

Misconstrued typing aside, all items will have to popped from Redis for `utils.create_batch`'s use to be justified.