Ticket #8901: 8901-r13323.diff

django/db/backends/postgresql/operations.py

@@ -54 +54 @@
         return '%s'
 
     def last_insert_id(self, cursor, table_name, pk_name):
-        cursor.execute("SELECT CURRVAL('\"%s_%s_seq\"')" % (table_name, pk_name))
+        # Use pg_get_serial_sequence to get the underlying sequence name
+        # from the table name and column name (available since PostgreSQL 8)
+        cursor.execute("SELECT CURRVAL(pg_get_serial_sequence('%s','%s'))" % (table_name, pk_name))
         return cursor.fetchone()[0]
 
     def no_limit_value(self):
@@ -90 +92 @@
             for sequence_info in sequences:
                 table_name = sequence_info['table']
                 column_name = sequence_info['column']
-                if column_name and len(column_name) > 0:
-                    sequence_name = '%s_%s_seq' % (table_name, column_name)
-                else:
-                    sequence_name = '%s_id_seq' % table_name
-                sql.append("%s setval('%s', 1, false);" % \
+                if not (column_name and len(column_name) > 0):
+                    # This will be the case if it's an m2m using an autogenerated
+                    # intermediate table (see BaseDatabaseIntrospection.sequence_list)
+                    column_name = 'id'
+                sql.append("%s setval(pg_get_serial_sequence('%s','%s'), 1, false);" % \
                     (style.SQL_KEYWORD('SELECT'),
-                    style.SQL_FIELD(self.quote_name(sequence_name)))
+                    style.SQL_TABLE(table_name),
+                    style.SQL_FIELD(column_name))
                 )
             return sql
         else:
@@ -110 +113 @@
             # Use `coalesce` to set the sequence for each model to the max pk value if there are records,
             # or 1 if there are none. Set the `is_called` property (the third argument to `setval`) to true
             # if there are records (as the max pk value is already in use), otherwise set it to false.
+            # Use pg_get_serial_sequence to get the underlying sequence name from the table name
+            # and column name (available since PostgreSQL 8)
+
             for f in model._meta.local_fields:
                 if isinstance(f, models.AutoField):
-                    output.append("%s setval('%s', coalesce(max(%s), 1), max(%s) %s null) %s %s;" % \
+                    output.append("%s setval(pg_get_serial_sequence('%s','%s'), coalesce(max(%s), 1), max(%s) %s null) %s %s;" % \
                         (style.SQL_KEYWORD('SELECT'),
-                        style.SQL_FIELD(qn('%s_%s_seq' % (model._meta.db_table, f.column))),
+                        style.SQL_TABLE(model._meta.db_table),
+                        style.SQL_FIELD(f.column),
                         style.SQL_FIELD(qn(f.column)),
                         style.SQL_FIELD(qn(f.column)),
                         style.SQL_KEYWORD('IS NOT'),
@@ -123 +130 @@
                     break # Only one AutoField is allowed per model, so don't bother continuing.
             for f in model._meta.many_to_many:
                 if not f.rel.through:
-                    output.append("%s setval('%s', coalesce(max(%s), 1), max(%s) %s null) %s %s;" % \
+                    output.append("%s setval(pg_get_serial_sequence('%s','%s'), coalesce(max(%s), 1), max(%s) %s null) %s %s;" % \
                         (style.SQL_KEYWORD('SELECT'),
-                        style.SQL_FIELD(qn('%s_id_seq' % f.m2m_db_table())),
+                        style.SQL_TABLE(model._meta.db_table),
+                        style.SQL_FIELD('id'),
                         style.SQL_FIELD(qn('id')),
                         style.SQL_FIELD(qn('id')),
                         style.SQL_KEYWORD('IS NOT'),

tests/regressiontests/backends/models.py

@@ -1 +1 @@
 from django.db import models
 from django.db import connection
 
+
 class Square(models.Model):
     root = models.IntegerField()
     square = models.PositiveIntegerField()
@@ -8 +9 @@
     def __unicode__(self):
         return "%s ** 2 == %s" % (self.root, self.square)
 
+
 class Person(models.Model):
     first_name = models.CharField(max_length=20)
     last_name = models.CharField(max_length=20)
@@ -15 +17 @@
     def __unicode__(self):
         return u'%s %s' % (self.first_name, self.last_name)
 
+
 class SchoolClass(models.Model):
     year = models.PositiveIntegerField()
     day = models.CharField(max_length=9, blank=True)
     last_updated = models.DateTimeField()
 
+
+class VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ(models.Model):
+    class Meta:
+        # We need to use a short actual table name or
+        # we hit issue #8548 which we're not testing!
+        verbose_name = 'model_with_long_table_name'
+    primary_key_is_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz = models.AutoField(primary_key=True)
+    m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz = models.ManyToManyField(Person,blank=True)
+
+
 qn = connection.ops.quote_name
 
 __test__ = {'API_TESTS': """

tests/regressiontests/backends/tests.py

@@ -1 +1 @@
 # -*- coding: utf-8 -*-
 # Unit and doctests for specific database backends.
 import datetime
-import models
 import unittest
+
+from django.conf import settings
+from django.core import management
+from django.core.management.color import no_style
 from django.db import backend, connection, DEFAULT_DB_ALIAS
 from django.db.backends.signals import connection_created
-from django.conf import settings
 from django.test import TestCase
 
+from regressiontests.backends import models
+
 class Callproc(unittest.TestCase):
 
     def test_dbms_session(self):
@@ -88 +92 @@
         self.assertRaises(Exception, cursor.executemany, query, [(1,2,3),])
         self.assertRaises(Exception, cursor.executemany, query, [(1,),])
 
+class LongNameTest(TestCase):
+    """Long primary keys and model names can result in a sequence name
+    that exceeds the database limits, which will result in truncation
+    on certain databases (e.g., Postgres). The backend needs to use
+    the correct sequence name in last_insert_id and other places, so
+    check it is. Refs #8901.
+    """
+
+    def test_sequence_name_length_limits_create(self):
+        """Test creation of model with long name and long pk name doesn't error. Ref #8901"""
+        models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create()
+
+    def test_sequence_name_length_limits_m2m(self):
+        """Test an m2m save of a model with a long name and a long m2m field name doesn't error as on Django >=1.2 this now uses object saves. Ref #8901"""
+        obj = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ.objects.create()
+        rel_obj = models.Person.objects.create(first_name='Django', last_name='Reinhardt')
+        obj.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.add(rel_obj)
+
+    def test_sequence_name_length_limits_flush(self):
+        """Test that sequence resetting as part of a flush with model with long name and long pk name doesn't error. Ref #8901"""
+        # A full flush is expensive to the full test, so we dig into the
+        # internals to generate the likely offending SQL and run it manually
+
+        # Some convenience aliases
+        VLM = models.VeryLongModelNameZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ
+        VLM_m2m = VLM.m2m_also_quite_long_zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz.through
+        tables = [
+            VLM._meta.db_table,
+            VLM_m2m._meta.db_table,
+        ]
+        sequences = [
+            {
+                'column': VLM._meta.pk.attname,
+                'table': VLM._meta.db_table
+            },
+        ]
+        cursor = connection.cursor()
+        for statement in connection.ops.sql_flush(no_style(), tables, sequences):
+            cursor.execute(statement)
+
 
 def connection_created_test(sender, **kwargs):
     print 'connection_created signal'