Handle Sequence in _process_multiparam_default_bind

Fixed issue where usage of an explicit :class:`.Sequence` would produce
inconsistent "inline" behavior for an :class:`.Insert` construct that
includes multiple values phrases; the first seq would be inline but
subsequent ones would be "pre-execute", leading to inconsistent sequence
ordering. The sequence expressions are now fully inline.

Fixes: #6361
Change-Id: Ie16794ec0e19979a7e6c8d1bef5716a9fc199889

diff --git a/doc/build/changelog/unreleased_14/6361.rst b/doc/build/changelog/unreleased_14/6361.rst
new file mode 100644 (file)
index 0000000..3b4b137
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6361.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, engine
+    :tickets: 6361
+
+    Fixed issue where usage of an explicit :class:`.Sequence` would produce
+    inconsistent "inline" behavior for an :class:`.Insert` construct that
+    includes multiple values phrases; the first seq would be inline but
+    subsequent ones would be "pre-execute", leading to inconsistent sequence
+    ordering. The sequence expressions are now fully inline.

diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index 5fa82bcd09c9e0e87f81cf17e69a0e5c731864be..25dc0d23228e815145f44a44fa2ea495df673763 100644 (file)
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -803,7 +803,6 @@ class _multiparam_column(elements.ColumnElement):
 
 
 def _process_multiparam_default_bind(compiler, stmt, c, index, kw):
-
     if not c.default:
         raise exc.CompileError(
             "INSERT value for column %s is explicitly rendered as a bound"
@@ -812,6 +811,16 @@ def _process_multiparam_default_bind(compiler, stmt, c, index, kw):
         )
     elif c.default.is_clause_element:
         return compiler.process(c.default.arg.self_group(), **kw)
+    elif c.default.is_sequence:
+        # these conditions would have been established
+        # by append_param_insert_(?:hasdefault|pk_returning|pk_no_returning)
+        # in order for us to be here, so these don't need to be
+        # checked
+        # assert compiler.dialect.supports_sequences and (
+        #    not c.default.optional
+        #    or not compiler.dialect.sequences_optional
+        # )
+        return compiler.process(c.default, **kw)
     else:
         col = _multiparam_column(c, index)
         if isinstance(stmt, dml.Insert):

diff --git a/test/sql/test_insert.py b/test/sql/test_insert.py
index 95a8d02a27465d7e6b5f8f005f9816a2040f1bd7..6c2a5d9557a53991c47e32752647c9adad28cfd5 100644 (file)
--- a/test/sql/test_insert.py
+++ b/test/sql/test_insert.py
@@ -384,6 +384,75 @@ class InsertTest(_InsertTestBase, fixtures.TablesTest, AssertsCompiledSQL):
             dialect=postgresql.dialect(),
         )
 
+    def test_insert_seq_pk_multi_values(self):
+        """test #6361"""
+
+        m = MetaData()
+
+        t1 = Table(
+            "t",
+            m,
+            Column("id", Integer, Sequence("id_seq"), primary_key=True),
+            Column("data", String),
+        )
+
+        stmt = t1.insert().values(
+            [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+        )
+
+        self.assert_compile(
+            stmt,
+            "INSERT INTO t (id, data) VALUES (nextval('id_seq'), "
+            "%(data_m0)s), (nextval('id_seq'), %(data_m1)s), "
+            "(nextval('id_seq'), %(data_m2)s)",
+            dialect=postgresql.dialect(),
+        )
+
+    def test_insert_seq_non_pk_multi_values(self):
+        """test #6361"""
+
+        m = MetaData()
+
+        t1 = Table(
+            "t",
+            m,
+            Column("id", Integer, primary_key=True),
+            Column("counter", Sequence("counter_seq")),
+            Column("data", String),
+        )
+
+        stmt = t1.insert().values(
+            [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+        )
+
+        self.assert_compile(
+            stmt,
+            "INSERT INTO t (counter, data) VALUES (nextval('counter_seq'), "
+            "%(data_m0)s), (nextval('counter_seq'), %(data_m1)s), "
+            "(nextval('counter_seq'), %(data_m2)s)",
+            dialect=postgresql.dialect(),
+        )
+
+    def test_insert_seq_pk_multi_values_seq_not_supported(self):
+        m = MetaData()
+
+        t1 = Table(
+            "t",
+            m,
+            Column("id", Integer, Sequence("id_seq"), primary_key=True),
+            Column("data", String),
+        )
+
+        stmt = t1.insert().values(
+            [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+        )
+
+        self.assert_compile(
+            stmt,
+            "INSERT INTO t (data) VALUES (?), (?), (?)",
+            dialect=sqlite.dialect(),
+        )
+
     def test_insert_from_select_cte_one(self):
         table1 = self.tables.mytable
 

diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index 3b25ba2ad0d8da02bfde1d863adb1e25361e30e3..d1d46afa3295192f3af534d90550516713c9a131 100644 (file)
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -204,6 +204,44 @@ class SequenceExecTest(fixtures.TestBase):
             else:
                 eq_(r.inserted_primary_key, (None,))
 
+    @testing.combinations(
+        ("implicit_returning",),
+        ("no_implicit_returning",),
+        ("explicit_returning", testing.requires.returning),
+        argnames="returning",
+    )
+    @testing.requires.multivalues_inserts
+    def test_seq_multivalues_inline(self, metadata, testing_engine, returning):
+        t1 = Table(
+            "t",
+            metadata,
+            Column("x", Integer, Sequence("my_seq"), primary_key=True),
+            Column("data", String(50)),
+        )
+
+        e = engines.testing_engine(
+            options={
+                "implicit_returning": returning != "no_implicit_returning"
+            }
+        )
+        metadata.create_all(e)
+        with e.begin() as conn:
+
+            stmt = t1.insert().values(
+                [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+            )
+            if returning == "explicit_returning":
+                stmt = stmt.returning(t1.c.x)
+
+            r = conn.execute(stmt)
+            if returning == "explicit_returning":
+                eq_(r.all(), [(1,), (2,), (3,)])
+
+            eq_(
+                conn.execute(t1.select().order_by(t1.c.x)).all(),
+                [(1, "d1"), (2, "d2"), (3, "d3")],
+            )
+
     @testing.requires.returning
     @testing.provide_metadata
     def test_inserted_pk_implicit_returning(self):