Fixed #34171 -- Fixed QuerySet.bulk_create() on fields with db_column in unique_fields/update_fields.

Bug in 0f6946495a8ec955b471ca1baaf408ceb53d4796.

Thanks Joshua Brooks for the report.

5 files changed, 46 insertions, 7 deletions

diff --git a/django/db/models/query.py b/django/db/models/query.py
index be8580f2c4..cf419cb8cf 100644
--- a/django/db/models/query.py
+++ b/django/db/models/query.py
@@ -720,7 +720,6 @@ class QuerySet(AltersData):
                     "Unique fields that can trigger the upsert must be provided."
                 )
             # Updating primary keys and non-concrete fields is forbidden.
-            update_fields = [self.model._meta.get_field(name) for name in update_fields]
             if any(not f.concrete or f.many_to_many for f in update_fields):
                 raise ValueError(
                     "bulk_create() can only be used with concrete fields in "
@@ -732,9 +731,6 @@ class QuerySet(AltersData):
                     "update_fields."
                 )
             if unique_fields:
-                unique_fields = [
-                    self.model._meta.get_field(name) for name in unique_fields
-                ]
                 if any(not f.concrete or f.many_to_many for f in unique_fields):
                     raise ValueError(
                         "bulk_create() can only be used with concrete fields "
@@ -786,8 +782,11 @@ class QuerySet(AltersData):
         if unique_fields:
             # Primary key is allowed in unique_fields.
             unique_fields = [
-                opts.pk.name if name == "pk" else name for name in unique_fields
+                self.model._meta.get_field(opts.pk.name if name == "pk" else name)
+                for name in unique_fields
             ]
+        if update_fields:
+            update_fields = [self.model._meta.get_field(name) for name in update_fields]
         on_conflict = self._check_bulk_create_options(
             ignore_conflicts,
             update_conflicts,
diff --git a/django/db/models/sql/compiler.py b/django/db/models/sql/compiler.py
index 0562a71dd1..caf36382b5 100644
--- a/django/db/models/sql/compiler.py
+++ b/django/db/models/sql/compiler.py
@@ -1725,8 +1725,8 @@ class SQLInsertCompiler(SQLCompiler):
         on_conflict_suffix_sql = self.connection.ops.on_conflict_suffix_sql(
             fields,
             self.query.on_conflict,
-            self.query.update_fields,
-            self.query.unique_fields,
+            (f.column for f in self.query.update_fields),
+            (f.column for f in self.query.unique_fields),
         )
         if (
             self.returning_fields
diff --git a/docs/releases/4.1.4.txt b/docs/releases/4.1.4.txt
index 7cdd2521ea..509135f3b2 100644
--- a/docs/releases/4.1.4.txt
+++ b/docs/releases/4.1.4.txt
@@ -23,3 +23,6 @@ Bugfixes
 
 * Fixed a bug in Django 4.1 that caused a crash of ``QuerySet.bulk_create()``
   with ``"pk"`` in ``unique_fields`` (:ticket:`34177`).
+
+* Fixed a bug in Django 4.1 that caused a crash of ``QuerySet.bulk_create()``
+  on fields with ``db_column`` (:ticket:`34171`).
diff --git a/tests/bulk_create/models.py b/tests/bulk_create/models.py
index 27abc416bd..8a21c7dfa1 100644
--- a/tests/bulk_create/models.py
+++ b/tests/bulk_create/models.py
@@ -69,6 +69,11 @@ class TwoFields(models.Model):
     name = models.CharField(max_length=15, null=True)
 
 
+class FieldsWithDbColumns(models.Model):
+    rank = models.IntegerField(unique=True, db_column="rAnK")
+    name = models.CharField(max_length=15, null=True, db_column="oTheRNaMe")
+
+
 class UpsertConflict(models.Model):
     number = models.IntegerField(unique=True)
     rank = models.IntegerField()
diff --git a/tests/bulk_create/tests.py b/tests/bulk_create/tests.py
index 6c490d1235..a5050c9b0b 100644
--- a/tests/bulk_create/tests.py
+++ b/tests/bulk_create/tests.py
@@ -21,6 +21,7 @@ from django.test import (
 from .models import (
     BigAutoFieldModel,
     Country,
+    FieldsWithDbColumns,
     NoFields,
     NullableFields,
     Pizzeria,
@@ -772,3 +773,34 @@ class BulkCreateTests(TestCase):
     @skipIfDBFeature("supports_update_conflicts_with_target")
     def test_update_conflicts_no_unique_fields(self):
         self._test_update_conflicts([])
+
+    @skipUnlessDBFeature(
+        "supports_update_conflicts", "supports_update_conflicts_with_target"
+    )
+    def test_update_conflicts_unique_fields_update_fields_db_column(self):
+        FieldsWithDbColumns.objects.bulk_create(
+            [
+                FieldsWithDbColumns(rank=1, name="a"),
+                FieldsWithDbColumns(rank=2, name="b"),
+            ]
+        )
+        self.assertEqual(FieldsWithDbColumns.objects.count(), 2)
+
+        conflicting_objects = [
+            FieldsWithDbColumns(rank=1, name="c"),
+            FieldsWithDbColumns(rank=2, name="d"),
+        ]
+        FieldsWithDbColumns.objects.bulk_create(
+            conflicting_objects,
+            update_conflicts=True,
+            unique_fields=["rank"],
+            update_fields=["name"],
+        )
+        self.assertEqual(FieldsWithDbColumns.objects.count(), 2)
+        self.assertCountEqual(
+            FieldsWithDbColumns.objects.values("rank", "name"),
+            [
+                {"rank": 1, "name": "c"},
+                {"rank": 2, "name": "d"},
+            ],
+        )