vassili/seitime-frappe
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(schema): drop unique constraint for deleted doctype fields (#36356)

Browse source 
* fix(schema): drop unique constraint and indexes for deleted doctype fields

* refactor(schema): rename a variable and remove commented code

* test: add test case for dropping unique constraint on field deletion from doctype

* fix(tests): prevent list mutation during iteration

* test(db): guard MariaDB-specific unique index test with db_type_is.MARIADB

* fix(schema): drop unique constraints and indexes for deleted fields on postgres

* fix(schema): make postgres unique cleanup idempotent for deleted fields

* fix(schema): make postgres unique cleanup idempotent on reload

* test: add test case for dropping unique constraint and index on field deletion for postgres

* fix(schema): make postgres unique cleanup idempotent
This commit is contained in:
Shllokkk 2026-02-20 20:15:32 +05:30 • committed by GitHub
parent 617e4a8d8f
commit 16e6c40d70
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 248 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
frappe/database/mariadb/schema.py
View file

@ -2,7 +2,7 @@ from pymysql.constants.ER import DUP_ENTRY
import frappe
from frappe import _
from frappe.database.schema import DBTable
from frappe.database.schema import DbColumn, DBTable
from frappe.utils.defaults import get_not_null_defaults
@ -96,6 +96,37 @@ class MariaDBTable(DBTable):
):
add_index_query.append("ADD INDEX `modified`(`modified`)")
# logic to drop unique constraint for fields deleted from a doctype
meta_columns = set(self.columns.keys())
db_columns = set(self.current_columns.keys())
for col in db_columns:
if (
col not in meta_columns
and col not in frappe.db.DEFAULT_COLUMNS
and col not in frappe.db.OPTIONAL_COLUMNS
):
has_unique = frappe.db.get_column_index(self.table_name, col, unique=True)
if not has_unique:
continue
current_col = self.current_columns.get(col)
deleted_col = DbColumn(
table=self,
fieldname=current_col.name,
fieldtype=current_col.type,
length=None,
default=None,
set_index=current_col.index,
options=None,
unique=False,
precision=None,
not_nullable=current_col.not_nullable,
)
self.drop_unique.append(deleted_col)
drop_index_query = []
for col in {*self.drop_index, *self.drop_unique}:

77
frappe/database/postgres/schema.py
View file

@ -1,6 +1,6 @@
import frappe
from frappe import _
from frappe.database.schema import DBTable, get_definition
from frappe.database.schema import DbColumn, DBTable, get_definition
from frappe.utils import cint, flt
from frappe.utils.defaults import get_not_null_defaults
@ -131,6 +131,50 @@ class PostgresTable(DBTable):
index_name=col.fieldname, table_name=self.table_name, field=col.fieldname
)
# logic to drop unique constraint for fields deleted from a doctype
meta_columns = set(self.columns.keys())
db_columns = set(self.current_columns.keys())
for col in db_columns:
if (
col not in meta_columns
and col not in frappe.db.DEFAULT_COLUMNS
and col not in frappe.db.OPTIONAL_COLUMNS
):
has_unique_index = frappe.db.sql(
"""
SELECT 1
FROM pg_indexes
WHERE tablename = %s
AND indexname IN (%s, %s)
LIMIT 1
""",
(
self.table_name,
f"{self.table_name}_{col}_key",
f"unique_{col}",
),
)
if not has_unique_index:
continue
current_col = self.current_columns.get(col)
deleted_col = DbColumn(
table=self,
fieldname=current_col.name,
fieldtype=current_col.type,
length=None,
default=None,
set_index=current_col.index,
options=None,
unique=False,
precision=None,
not_nullable=current_col.not_nullable,
)
self.drop_unique.append(deleted_col)
drop_contraint_query = ""
for col in self.drop_index:
# primary key
@ -141,8 +185,35 @@ class PostgresTable(DBTable):
for col in self.drop_unique:
# primary key
if col.fieldname != "name":
# if index key exists
drop_contraint_query += f'DROP INDEX IF EXISTS "unique_{col.fieldname}" ;'
# drop unique constraint first if exists which automatically drops the underlying index also
unique_constraint_exists = frappe.db.sql(
"""
SELECT 1
FROM pg_constraint
WHERE conname = %s
""",
(f"{self.table_name}_{col.fieldname}_key",),
)
if unique_constraint_exists:
drop_contraint_query += f'ALTER TABLE "{self.table_name}" DROP CONSTRAINT IF EXISTS "{self.table_name}_{col.fieldname}_key" ;'
# drop the unique index backed by no constraint directly
unique_index_exists = frappe.db.sql(
"""
SELECT 1
FROM pg_indexes
WHERE tablename = %s
AND indexname = %s
""",
(
self.table_name,
f"unique_{col.fieldname}",
),
)
if unique_index_exists:
drop_contraint_query += f'DROP INDEX IF EXISTS "unique_{col.fieldname}" ;'
change_nullability = []
for col in self.change_nullability:

142
frappe/tests/test_query.py
View file

@ -2308,6 +2308,148 @@ class TestQuery(IntegrationTestCase):
self.assertEqual(engine._get_ifnull_fallback("Patch Log", "skipped"), "0")
self.assertEqual(engine._get_ifnull_fallback("Patch Log", "patch"), "''")
@run_only_if(db_type_is.MARIADB)
def test_drop_unique_constraint_for_deleted_fields_mariadb(self):
trial_dt = new_doctype(
"Trial Doctype",
fields=[
{
"fieldname": "field_one",
"fieldtype": "Data",
"label": "Field One",
},
{
"fieldname": "field_two",
"fieldtype": "Data",
"label": "Field Two",
"unique": 1,
},
],
)
trial_dt.insert(ignore_if_duplicate=True)
indexes = frappe.db.get_column_index("tabTrial Doctype", "field_two", unique=True)
self.assertTrue(indexes)
field_to_remove = None
for field in trial_dt.fields:
if field.fieldname == "field_two":
field_to_remove = field
break
trial_dt.fields.remove(field_to_remove)
trial_dt.save()
indexes = frappe.db.get_column_index("tabTrial Doctype", "field_two", unique=True)
self.assertFalse(indexes)
@run_only_if(db_type_is.POSTGRES)
def test_drop_unique_constraint_and_indexes_for_deleted_fields_postgres(self):
# test for unique index backed by constraint at field creation time
trial_dt = new_doctype(
"Trial Doctype",
fields=[
{
"fieldname": "field_one",
"fieldtype": "Data",
"label": "Field One",
},
{
"fieldname": "field_two",
"fieldtype": "Data",
"label": "Field Two",
"unique": 1,
},
],
)
trial_dt.insert(ignore_if_duplicate=True)
index_exists = frappe.db.sql(
"""
SELECT 1
FROM pg_indexes
WHERE tablename = %s
AND indexname = %s
""",
(
f"tab{trial_dt.name}",
f"tab{trial_dt.name}_field_two_key",
),
)
self.assertTrue(index_exists)
field_to_remove = None
for field in trial_dt.fields:
if field.fieldname == "field_two":
field_to_remove = field
break
trial_dt.fields.remove(field_to_remove)
trial_dt.save()
index_exists = frappe.db.sql(
"""
SELECT 1
FROM pg_indexes
WHERE tablename = %s
AND indexname = %s
""",
(
f"tab{trial_dt.name}",
f"tab{trial_dt.name}_field_two_key",
),
)
self.assertFalse(index_exists)
# test for unique index backed by no constraint created at field alteration post creation
for field in trial_dt.fields:
if field.fieldname == "field_one":
field.unique = 1
trial_dt.save()
index_exists = frappe.db.sql(
"""
SELECT 1
FROM pg_indexes
WHERE tablename = %s
AND indexname = %s
""",
(
f"tab{trial_dt.name}",
"unique_field_one",
),
)
self.assertTrue(index_exists)
field_to_remove = None
for field in trial_dt.fields:
if field.fieldname == "field_one":
field_to_remove = field
break
trial_dt.fields.remove(field_to_remove)
trial_dt.save()
index_exists = frappe.db.sql(
"""
SELECT 1
FROM pg_indexes
WHERE tablename = %s
AND indexname = %s
""",
(
f"tab{trial_dt.name}",
"unique_field_one",
),
)
self.assertFalse(index_exists)
# This function is used as a permission query condition hook
def test_permission_hook_condition(user):