explicitly fetch inserted pk for values(pkcol=None)

Altered the compilation mechanics of the :class:`.Insert` construct such
that the "autoincrement primary key" column value will be fetched via
``cursor.lastrowid`` or RETURNING even if present in the parameter set or
within the :meth:`.Insert.values` method as a plain bound value, for
single-row INSERT statements on specific backends that are known to
generate autoincrementing values even when explicit NULL is passed. This
restores a behavior that was in the 1.3 series for both the use case of
separate parameter set as well as :meth:`.Insert.values`. In 1.4, the
parameter set behavior unintentionally changed to no longer do this, but
the :meth:`.Insert.values` method would still fetch autoincrement values up
until 1.4.21 where :ticket:`6770` changed the behavior yet again again
unintentionally as this use case was never covered.

The behavior is now defined as "working" to suit the case where databases
such as SQLite, MySQL and MariaDB will ignore an explicit NULL primary key
value and nonetheless invoke an autoincrement generator.

Fixes: #7998
Change-Id: I5d4105a14217945f87fbe9a6f2a3c87f6ef20529

5 files changed, 77 insertions, 7 deletions

diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index 62bdb9535..420fcfdfd 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -2359,6 +2359,7 @@ class MySQLDialect(default.DefaultDialect):
     supports_sane_rowcount = True
     supports_sane_multi_rowcount = False
     supports_multivalues_insert = True
+    insert_null_pk_still_autoincrements = True
 
     supports_comments = True
     inline_comments = True
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index e2349295c..ab525581a 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1836,6 +1836,7 @@ class SQLiteDialect(default.DefaultDialect):
     supports_multivalues_insert = True
     tuple_in_values = True
     supports_statement_cache = True
+    insert_null_pk_still_autoincrements = True
 
     default_paramstyle = "qmark"
     execution_ctx_cls = SQLiteExecutionContext
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 653b73a32..926ad2077 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -135,6 +135,7 @@ class DefaultDialect(Dialect):
     preexecute_autoincrement_sequences = False
     supports_identity_columns = False
     postfetch_lastrowid = True
+    insert_null_pk_still_autoincrements = False
     implicit_returning = False
     full_returning = False
     insert_executemany_returning = False
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 19435b59c..f774028f2 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1560,14 +1560,36 @@ class SQLCompiler(Compiled):
             for col in table.primary_key
         ]
 
+        autoinc_getter = None
         autoinc_col = table._autoincrement_column
         if autoinc_col is not None:
             # apply type post processors to the lastrowid
-            proc = autoinc_col.type._cached_result_processor(
+            lastrowid_processor = autoinc_col.type._cached_result_processor(
                 self.dialect, None
             )
+            autoinc_key = param_key_getter(autoinc_col)
+
+            # if a bind value is present for the autoincrement column
+            # in the parameters, we need to do the logic dictated by
+            # #7998; honor a non-None user-passed parameter over lastrowid.
+            # previously in the 1.4 series we weren't fetching lastrowid
+            # at all if the key were present in the parameters
+            if autoinc_key in self.binds:
+
+                def autoinc_getter(lastrowid, parameters):
+                    param_value = parameters.get(autoinc_key, lastrowid)
+                    if param_value is not None:
+                        # they supplied non-None parameter, use that.
+                        # SQLite at least is observed to return the wrong
+                        # cursor.lastrowid for INSERT..ON CONFLICT so it
+                        # can't be used in all cases
+                        return param_value
+                    else:
+                        # use lastrowid
+                        return lastrowid
+
         else:
-            proc = None
+            lastrowid_processor = None
 
         row_fn = result.result_tuple([col.key for col in table.primary_key])
 
@@ -1578,14 +1600,20 @@ class SQLCompiler(Compiled):
             that were sent along with the INSERT.
 
             """
-            if proc is not None:
-                lastrowid = proc(lastrowid)
+            if lastrowid_processor is not None:
+                lastrowid = lastrowid_processor(lastrowid)
 
             if lastrowid is None:
                 return row_fn(getter(parameters) for getter, col in getters)
             else:
                 return row_fn(
-                    lastrowid if col is autoinc_col else getter(parameters)
+                    (
+                        autoinc_getter(lastrowid, parameters)
+                        if autoinc_getter
+                        else lastrowid
+                    )
+                    if col is autoinc_col
+                    else getter(parameters)
                     for getter, col in getters
                 )
 
diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index d3329c391..6a5785043 100644
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -528,6 +528,17 @@ def _scan_cols(
     else:
         cols = stmt.table.columns
 
+    if compile_state.isinsert and not compile_state._has_multi_parameters:
+        # new rules for #7998.  fetch lastrowid or implicit returning
+        # for autoincrement column even if parameter is NULL, for DBs that
+        # override NULL param for primary key (sqlite, mysql/mariadb)
+        autoincrement_col = stmt.table._autoincrement_column
+        insert_null_pk_still_autoincrements = (
+            compiler.dialect.insert_null_pk_still_autoincrements
+        )
+    else:
+        autoincrement_col = insert_null_pk_still_autoincrements = None
+
     for c in cols:
         # scan through every column in the target table
 
@@ -547,6 +558,8 @@ def _scan_cols(
                 implicit_returning,
                 implicit_return_defaults,
                 values,
+                autoincrement_col,
+                insert_null_pk_still_autoincrements,
                 kw,
             )
 
@@ -626,6 +639,8 @@ def _append_param_parameter(
     implicit_returning,
     implicit_return_defaults,
     values,
+    autoincrement_col,
+    insert_null_pk_still_autoincrements,
     kw,
 ):
     value = parameters.pop(col_key)
@@ -635,6 +650,19 @@ def _append_param_parameter(
     )
 
     if coercions._is_literal(value):
+
+        if (
+            insert_null_pk_still_autoincrements
+            and c.primary_key
+            and c is autoincrement_col
+        ):
+            # support use case for #7998, fetch autoincrement cols
+            # even if value was given
+            if implicit_returning:
+                compiler.implicit_returning.append(c)
+            elif compiler.dialect.postfetch_lastrowid:
+                compiler.postfetch_lastrowid = True
+
         value = _create_bind_param(
             compiler,
             c,
@@ -646,6 +674,19 @@ def _append_param_parameter(
             **kw,
         )
     elif value._is_bind_parameter:
+        if (
+            insert_null_pk_still_autoincrements
+            and value.value is None
+            and c.primary_key
+            and c is autoincrement_col
+        ):
+            # support use case for #7998, fetch autoincrement cols
+            # even if value was given
+            if implicit_returning:
+                compiler.implicit_returning.append(c)
+            elif compiler.dialect.postfetch_lastrowid:
+                compiler.postfetch_lastrowid = True
+
         value = _handle_values_anonymous_param(
             compiler,
             c,
@@ -1244,7 +1285,6 @@ def _get_returning_modifiers(compiler, stmt, compile_state, toplevel):
         and not stmt._returning
         and not compile_state._has_multi_parameters
     )
-
     implicit_returning = (
         need_pks
         and compiler.dialect.implicit_returning
@@ -1271,7 +1311,6 @@ def _get_returning_modifiers(compiler, stmt, compile_state, toplevel):
             implicit_return_defaults = set(stmt._return_defaults_columns)
 
     postfetch_lastrowid = need_pks and compiler.dialect.postfetch_lastrowid
-
     return (
         need_pks,
         implicit_returning,