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, 58 insertions, 0 deletions

diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index 2d84ab676..3f74bdbcc 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -87,6 +87,16 @@ table_a_2_bs = Table(
 
 table_b = Table("b", meta, Column("a", Integer), Column("b", Integer))
 
+table_b_b = Table(
+    "b_b",
+    meta,
+    Column("a", Integer),
+    Column("b", Integer),
+    Column("c", Integer),
+    Column("d", Integer),
+    Column("e", Integer),
+)
+
 table_c = Table("c", meta, Column("x", Integer), Column("y", Integer))
 
 table_d = Table("d", meta, Column("y", Integer), Column("z", Integer))
@@ -711,6 +721,54 @@ class CoreFixtures(object):
 
     fixtures.append(_statements_w_anonymous_col_names)
 
+    def _update_dml_w_dicts():
+        return (
+            table_b_b.update().values(
+                {
+                    table_b_b.c.a: 5,
+                    table_b_b.c.b: 5,
+                    table_b_b.c.c: 5,
+                    table_b_b.c.d: 5,
+                }
+            ),
+            # equivalent, but testing dictionary insert ordering as cache key
+            # / compare
+            table_b_b.update().values(
+                {
+                    table_b_b.c.a: 5,
+                    table_b_b.c.c: 5,
+                    table_b_b.c.b: 5,
+                    table_b_b.c.d: 5,
+                }
+            ),
+            table_b_b.update().values(
+                {table_b_b.c.a: 5, table_b_b.c.b: 5, "c": 5, table_b_b.c.d: 5}
+            ),
+            table_b_b.update().values(
+                {
+                    table_b_b.c.a: 5,
+                    table_b_b.c.b: 5,
+                    table_b_b.c.c: 5,
+                    table_b_b.c.d: 5,
+                    table_b_b.c.e: 10,
+                }
+            ),
+            table_b_b.update()
+            .values(
+                {
+                    table_b_b.c.a: 5,
+                    table_b_b.c.b: 5,
+                    table_b_b.c.c: 5,
+                    table_b_b.c.d: 5,
+                    table_b_b.c.e: 10,
+                }
+            )
+            .where(table_b_b.c.c > 10),
+        )
+
+    if util.py37:
+        fixtures.append(_update_dml_w_dicts)
+
 
 class CacheKeyFixture(object):
     def _run_cache_key_fixture(self, fixture, compare_values):