Do not rely on INSERT returning in the same order as VALUES

There is no guarantee by SQL standard that the order of the returned values of
INSERT are the same as the order of VALUES.
Also there is neither any guarantee that the generated id follow the order of
VALUES but the ORM relies on such order to keep records ordered in the same
order as the user insert them (and also for sequence order with NULL value).
So for database that supports multirow insert, we need to set explicitly the
value for the column id to ensure to preserve the insert order.

Closes #12220
This commit is contained in:
Cédric Krier committed 2023-04-28 19:04:02 +02:00
1 parent b4d78247cf
commit 8039dfd5a4
2 files changed
+86 -20

No files matched your search

+30 -17
View File
@@ -761,8 +761,7 @@ class ModelSQL(ModelStorage):
vlist = [v.copy() for v in vlist]
def db_insert(columns, vlist, fields):
if (transaction.database.has_multirow_insert()
and transaction.database.has_returning()):
if transaction.database.has_multirow_insert():
vlist = (
s for s in grouped_slice(
vlist, in_max // (len(fields) or 1)))
@@ -771,23 +770,37 @@ class ModelSQL(ModelStorage):
for values in vlist:
values = list(values)
cols = list(columns)
try:
if transaction.database.has_returning():
cursor.execute(*table.insert(
columns, values, [table.id]))
yield from (r[0] for r in cursor)
else:
id_new = transaction.database.nextid(
transaction.connection, cls._table)
if id_new:
if 'id' not in {c.name for c in columns}:
columns.append(table.id)
values[0].append(id_new)
cursor.execute(*table.insert(columns, values))
if len(values) > 1:
ids = transaction.database.nextid(
transaction.connection, cls._table,
count=len(values))
if ids is not None:
assert len(ids) == len(values)
cols.append(table.id)
for val, id in zip(values, ids):
val.append(id)
cursor.execute(*table.insert(cols, values))
yield from ids
continue
for i, val in enumerate(values):
if transaction.database.has_returning():
cursor.execute(*table.insert(
cols, [val], [table.id]))
yield from (r[0] for r in cursor)
else:
cursor.execute(*table.insert(columns, values))
id_new = transaction.database.lastid(cursor)
yield id_new
id_new = transaction.database.nextid(
transaction.connection, cls._table)
if id_new:
if i == 0:
cols.append(table.id)
val.append(id_new)
cursor.execute(*table.insert(cols, [val]))
else:
cursor.execute(*table.insert(cols, [val]))
id_new = transaction.database.lastid(cursor)
yield id_new
except (
backend.DatabaseIntegrityError,
backend.DatabaseDataError
+56 -3
View File
@@ -366,6 +366,34 @@ class ModelSQLTestCase(unittest.TestCase):
ModelsqlTimestamp.delete([record])
transaction.commit()
@with_transaction()
def test_create(self):
"Test create record"
pool = Pool()
Model = pool.get('test.modelsql.create')
m1, = Model.create([{
'char': "Value 1",
}])
self.assertTrue(m1.id)
self.assertEqual(m1.char, "Value 1")
def test_create_without_returning(self):
"Test create record without returning"
with patch.object(backend.Database, 'has_returning') as returning:
returning.return_value = False
with Transaction().start(DB_NAME, USER, context=CONTEXT):
pool = Pool()
Model = pool.get('test.modelsql.create')
m1, = Model.create([{
'char': "Value 1",
}])
self.assertTrue(m1.id)
self.assertEqual(m1.char, "Value 1")
@with_transaction()
def test_create_many_records(self):
"Test create many records"
@@ -390,9 +418,10 @@ class ModelSQLTestCase(unittest.TestCase):
self.assertEqual(baz.char, "Baz")
self.assertEqual(baz.integer, None)
def test_create_no_returning(self):
with patch.object(backend.Database, 'has_returning') as returning:
returning.return_value = False
def test_create_many_records_without_multirow_insert(self):
"Test create many records without multirow insert"
with patch.object(backend.Database, 'has_multirow_insert') as multirow:
multirow.return_value = False
with Transaction().start(DB_NAME, USER, context=CONTEXT):
pool = Pool()
Model = pool.get('test.modelsql.create')
@@ -407,6 +436,30 @@ class ModelSQLTestCase(unittest.TestCase):
self.assertEqual(m1.char, "Value 1")
self.assertTrue(m2.id)
self.assertEqual(m2.char, "Value 2")
self.assertLess(m1.id, m2.id)
def test_create_many_without_returning_and_multirow_insert(self):
"Test create record without returning an multirow insert"
with patch.object(backend.Database, 'has_returning') as returning, \
patch.object(
backend.Database, 'has_multirow_insert') as multirow:
returning.return_value = False
multirow.return_value = False
with Transaction().start(DB_NAME, USER, context=CONTEXT):
pool = Pool()
Model = pool.get('test.modelsql.create')
m1, m2 = Model.create([{
'char': "Value 1",
}, {
'char': "Value 2",
}])
self.assertTrue(m1.id)
self.assertEqual(m1.char, "Value 1")
self.assertTrue(m2.id)
self.assertEqual(m2.char, "Value 2")
self.assertLess(m1.id, m2.id)
@with_transaction()
def test_create_field_set(self):