[IMP] util/pg: serialize queries upon DeadlockDetected

As recently noted in odoo/upgrade#5753, it is
possible that `parallel_execute` generates concurrency issues.
For instance, if a `DELETE` query is run in `//` on a table with a
self-referencing foreign key, based on the actual records in the db, the
following error may be encountered:

```
Traceback (most recent call last):
  File "/home/odoo/src/odoo/15.0/odoo/service/server.py", line 1260, in preload_registries
    registry = Registry.new(dbname, update_module=update_module)
  File "/home/odoo/src/odoo/15.0/odoo/modules/registry.py", line 87, in new
    odoo.modules.load_modules(registry, force_demo, status, update_module)
  File "/home/odoo/src/odoo/15.0/odoo/modules/loading.py", line 415, in load_modules
    loaded_modules, processed_modules = load_module_graph(
  File "/home/odoo/src/odoo/15.0/odoo/modules/loading.py", line 174, in load_module_graph
    migrations.migrate_module(package, 'pre')
  File "/home/odoo/src/odoo/15.0/odoo/modules/migration.py", line 186, in migrate_module
    migrate(self.cr, installed_version)
  File "/tmp/tmpjgh5os94/migrations/base/0.0.0/pre-clean-assets.py", line 7, in migrate
    util.parallel_execute(
  File "/tmp/tmpjgh5os94/migrations/util/pg.py", line 103, in parallel_execute
    return sum(
  File "/tmp/tmpjgh5os94/migrations/util/misc.py", line 215, in log_progress
    for i, e in enumerate(it, 1):
  File "/usr/lib/python3.8/concurrent/futures/_base.py", line 619, in result_iterator
    yield fs.pop().result()
  File "/usr/lib/python3.8/concurrent/futures/_base.py", line 437, in result
    return self.__get_result()
  File "/usr/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
    raise self._exception
  File "/usr/lib/python3.8/concurrent/futures/thread.py", line 57, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/tmp/tmpjgh5os94/migrations/util/pg.py", line 95, in execute
    cr.execute(query)
  File "<decorator-gen-5>", line 2, in execute
  File "/home/odoo/src/odoo/15.0/odoo/sql_db.py", line 90, in check
    return f(self, *args, **kwargs)
  File "/home/odoo/src/odoo/15.0/odoo/sql_db.py", line 311, in execute
    res = self._obj.execute(query, params)
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 1418247 waits for ShareLock on transaction 21800495; blocked by process 1418237.
Process 1418237 waits for ShareLock on transaction 21800496; blocked by process 1418247.
HINT:  See server log for query details.
CONTEXT:  while updating tuple (450,6) in relation "ir_attachment"
SQL statement "UPDATE ONLY "public"."ir_attachment" SET "doc_next_name_id" = NULL WHERE $1 OPERATOR(pg_catalog.=) "doc_next_name_id""
```

In the case of similar DeadlockDetected issues, the solution we have
adopted has always been to run the query (or the exploded query)
sequentially.

This commit implements a mechanism to automatically collect queries that
fail due to DeadlockDetected issues and rerun them sequentially with no
further failure control.

Some remarks:

* Potentially other errorcodes for other concurrent errors may be added
  in the future.
* `-2` was chosen as the error code, because `-1` is already used up by
  `cr.rowcount` when the number of affected rows can't be fetched (e.g.
  `CREATED INDEX` queries)

Part-of: #49
Co-authored-by: Carsten Wolff (cawo) <[email protected]>

Author: Pirols

src/util/pg.py

@@ -14,7 +14,8 @@
 from multiprocessing import cpu_count
 
 try:
-    from concurrent.futures import ThreadPoolExecutor
+    from concurrent.futures import ThreadPoolExecutor  # noqa: I001
+    import concurrent
 except ImportError:
     ThreadPoolExecutor = None
 
@@ -29,7 +30,7 @@
     pass
 
 import psycopg2
-from psycopg2 import sql
+from psycopg2 import errorcodes, sql
 
 try:
     from odoo.sql_db import db_connect
@@ -117,17 +118,34 @@ def execute(query):
 
         cr.commit()
 
+        CONCURRENCY_ERRORCODES = {errorcodes.DEADLOCK_DETECTED}
+        failed_queries = []
+        tot_cnt = 0
         with ThreadPoolExecutor(max_workers=max_workers) as executor:
-            return sum(
-                log_progress(
-                    executor.map(execute, queries),
-                    logger,
-                    qualifier="queries",
-                    size=len(queries),
-                    estimate=False,
-                    log_hundred_percent=True,
-                )
-            )
+            future_queries = {executor.submit(execute, q): q for q in queries}
+            for future in log_progress(
+                concurrent.futures.as_completed(future_queries),
+                logger,
+                qualifier="queries",
+                size=len(queries),
+                estimate=False,
+                log_hundred_percent=True,
+            ):
+                try:
+                    tot_cnt += future.result()
+                except psycopg2.OperationalError as exc:
+                    if exc.pgcode not in CONCURRENCY_ERRORCODES:
+                        raise
+
+                    # to be retried without concurrency
+                    failed_queries.append(future_queries[future])
+
+        if failed_queries:
+            logger.warning("Serialize queries that failed due to concurrency issues")
+            tot_cnt += _parallel_execute_serial(cr, failed_queries, logger=logger)
+            cr.commit()
+
+        return tot_cnt
 
 else:
     _parallel_execute_threaded = _parallel_execute_serial