fix: patch to delete duplicate uniq indexes

frappe/patches.txt

@@ -99,7 +99,7 @@ frappe.patches.v12_0.delete_feedback_request_if_exists #1
 frappe.patches.v12_0.rename_events_repeat_on
 frappe.patches.v12_0.fix_public_private_files
 frappe.patches.v12_0.move_email_and_phone_to_child_table
-frappe.patches.v12_0.delete_duplicate_indexes
+frappe.patches.v12_0.delete_duplicate_indexes  # 2022-12-15
 frappe.patches.v12_0.set_default_incoming_email_port
 frappe.patches.v12_0.update_global_search
 frappe.patches.v12_0.setup_tags

frappe/patches/v12_0/delete_duplicate_indexes.py

@@ -15,39 +15,41 @@ def execute():
 		indexes_to_keep_map = frappe._dict()
 		indexes_to_delete = []
 		index_info = frappe.db.sql(
-			"""
-			SELECT
-				column_name,
-				index_name,
-				non_unique
-			FROM information_schema.STATISTICS
-			WHERE table_name=%s
-			AND column_name!='name'
-			AND non_unique=0
-			ORDER BY index_name;
-		""",
-			table,
+			f"""SHOW INDEX FROM `{table}`
+				WHERE Seq_in_index = 1
+				AND Non_unique=0""",
 			as_dict=1,
 		)
 
 		for index in index_info:
-			if not indexes_to_keep_map.get(index.column_name):
-				indexes_to_keep_map[index.column_name] = index
+			if not indexes_to_keep_map.get(index.Column_name):
+				indexes_to_keep_map[index.Column_name] = index
 			else:
-				indexes_to_delete.append(index.index_name)
+				indexes_to_delete.append(index.Key_name)
+
 		if indexes_to_delete:
 			final_deletion_map[table] = indexes_to_delete
 
-	# build drop index query
-	for (table_name, index_list) in final_deletion_map.items():
-		query_list = []
-		alter_query = f"ALTER TABLE `{table_name}`"
-
+	for table_name, index_list in final_deletion_map.items():
 		for index in index_list:
-			query_list.append(f"{alter_query} DROP INDEX `{index}`")
-
-		for query in query_list:
 			try:
-				frappe.db.sql(query)
-			except frappe.db.InternalError:
-				pass
+				if is_clustered_index(table_name, index):
+					continue
+				frappe.db.sql_ddl(f"ALTER TABLE `{table_name}` DROP INDEX `{index}`")
+			except Exception as e:
+				frappe.log_error("Failed to drop index")
+				print(f"x Failed to drop index {index} from {table_name}\n {str(e)}")
+			else:
+				print(f"✓ dropped {index} index from {table}")
+
+
+def is_clustered_index(table, index_name):
+	return bool(
+		frappe.db.sql(
+			f"""SHOW INDEX FROM `{table}`
+			WHERE Key_name = "{index_name}"
+				AND Seq_in_index = 2
+			""",
+			as_dict=True,
+		)
+	)