Convert bulk update/delete to new execution model

This reorganizes the BulkUD model in sqlalchemy.orm.persistence
to be based on the CompileState concept and to allow plain
update() / delete() to be passed to session.execute() where
the ORM synchronize session logic will take place.
Also gets "synchronize_session='fetch'" working with horizontal
sharding.

Adding a few more result.scalar_one() types of methods
as scalar_one() seems like what is normally desired.

Fixes: #5160
Change-Id: I8001ebdad089da34119eb459709731ba6c0ba975

1 files changed, 43 insertions, 15 deletions

diff --git a/lib/sqlalchemy/sql/traversals.py b/lib/sqlalchemy/sql/traversals.py
index 388097e45..68281f33d 100644
--- a/lib/sqlalchemy/sql/traversals.py
+++ b/lib/sqlalchemy/sql/traversals.py
@@ -10,6 +10,7 @@ from .. import util
 from ..inspection import inspect
 from ..util import collections_abc
 from ..util import HasMemoized
+from ..util import py37
 
 SKIP_TRAVERSE = util.symbol("skip_traverse")
 COMPARE_FAILED = False
@@ -562,23 +563,38 @@ class _CacheKey(ExtendedInternalTraversal):
         )
 
     def visit_dml_values(self, attrname, obj, parent, anon_map, bindparams):
+        if py37:
+            # in py37 we can assume two dictionaries created in the same
+            # insert ordering will retain that sorting
+            return (
+                attrname,
+                tuple(
+                    (
+                        k._gen_cache_key(anon_map, bindparams)
+                        if hasattr(k, "__clause_element__")
+                        else k,
+                        obj[k]._gen_cache_key(anon_map, bindparams),
+                    )
+                    for k in obj
+                ),
+            )
+        else:
+            expr_values = {k for k in obj if hasattr(k, "__clause_element__")}
+            if expr_values:
+                # expr values can't be sorted deterministically right now,
+                # so no cache
+                anon_map[NO_CACHE] = True
+                return ()
 
-        expr_values = {k for k in obj if hasattr(k, "__clause_element__")}
-        if expr_values:
-            # expr values can't be sorted deterministically right now,
-            # so no cache
-            anon_map[NO_CACHE] = True
-            return ()
-
-        str_values = expr_values.symmetric_difference(obj)
+            str_values = expr_values.symmetric_difference(obj)
 
-        return (
-            attrname,
-            tuple(
-                (k, obj[k]._gen_cache_key(anon_map, bindparams))
-                for k in sorted(str_values)
-            ),
-        )
+            return (
+                attrname,
+                tuple(
+                    (k, obj[k]._gen_cache_key(anon_map, bindparams))
+                    for k in sorted(str_values)
+                ),
+            )
 
     def visit_dml_multi_values(
         self, attrname, obj, parent, anon_map, bindparams
@@ -1130,6 +1146,18 @@ class TraversalComparatorStrategy(InternalTraversal, util.MemoizedSlots):
             for lv, rv in zip(left, right):
                 if not self._compare_dml_values_or_ce(lv, rv, **kw):
                     return COMPARE_FAILED
+        elif isinstance(right, collections_abc.Sequence):
+            return COMPARE_FAILED
+        elif py37:
+            # dictionaries guaranteed to support insert ordering in
+            # py37 so that we can compare the keys in order.  without
+            # this, we can't compare SQL expression keys because we don't
+            # know which key is which
+            for (lk, lv), (rk, rv) in zip(left.items(), right.items()):
+                if not self._compare_dml_values_or_ce(lk, rk, **kw):
+                    return COMPARE_FAILED
+                if not self._compare_dml_values_or_ce(lv, rv, **kw):
+                    return COMPARE_FAILED
         else:
             for lk in left:
                 lv = left[lk]