refactor(stock): convert get_alternative_items UNION to ORM

Replace the raw UNION of forward/two-way alternative-item matches with two
frappe.get_all calls, order-preserving dedup (dict.fromkeys) and Python
pagination. Each leg is bounded to start+page_len rows so the per-keystroke
search round trip stays small (the original bounded with LIMIT/OFFSET);
ItemAlternative forbids duplicate (item_code, alternative_item_code) pairs, so
each leg is internally distinct and that bound is exact. Same result on
MariaDB; valid under Postgres.

Tests: both-direction dedup, txt filtering, pagination, bounded-and-exact
page-walk reconstruction, and case-insensitive (ILIKE-on-Postgres) matching.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

erpnext/stock/doctype/item_alternative/item_alternative.py

@@ -7,6 +7,7 @@ from typing import Any
 import frappe
 from frappe import _
 from frappe.model.document import Document
+from frappe.utils import cint
 
 
 class ItemAlternative(Document):
@@ -86,13 +87,24 @@ class ItemAlternative(Document):
 @frappe.whitelist()
 @frappe.validate_and_sanitize_search_inputs
 def get_alternative_items(doctype: Any, txt: str, searchfield: Any, start: int, page_len: int, filters: dict):
-	return frappe.db.sql(
-		f""" (select alternative_item_code from `tabItem Alternative`
-			where item_code = %(item_code)s and alternative_item_code like %(txt)s)
-		union
-			(select item_code from `tabItem Alternative`
-			where alternative_item_code = %(item_code)s and item_code like %(txt)s
-			and two_way = 1) limit {page_len} offset {start}
-		""",
-		{"item_code": filters.get("item_code"), "txt": "%" + txt + "%"},
+	item_code = filters.get("item_code")
+	search = f"%{txt}%"
+	# each leg has distinct values (validate_duplicate), so start+page_len rows per leg suffice
+	limit = cint(start) + cint(page_len)
+
+	alternatives = frappe.get_all(
+		"Item Alternative",
+		filters={"item_code": item_code, "alternative_item_code": ["like", search]},
+		pluck="alternative_item_code",
+		limit=limit,
 	)
+	alternatives += frappe.get_all(
+		"Item Alternative",
+		filters={"alternative_item_code": item_code, "item_code": ["like", search], "two_way": 1},
+		pluck="item_code",
+		limit=limit,
+	)
+
+	# union (dedupe, preserve order) + paginate
+	unique_items = list(dict.fromkeys(alternatives))
+	return [[item] for item in unique_items[start : start + page_len]]

erpnext/stock/doctype/item_alternative/test_item_alternative.py

@@ -2,7 +2,7 @@
 # See license.txt
 
 import frappe
-from frappe.utils import flt
+from frappe.utils import flt, random_string
 
 from erpnext.controllers.subcontracting_controller import make_rm_stock_entry
 from erpnext.controllers.tests.test_subcontracting_controller import (
@@ -14,6 +14,7 @@ from erpnext.manufacturing.doctype.production_plan.test_production_plan import m
 from erpnext.manufacturing.doctype.work_order.mapper import make_stock_entry
 from erpnext.manufacturing.doctype.work_order.test_work_order import make_wo_order_test_record
 from erpnext.stock.doctype.item.test_item import create_item
+from erpnext.stock.doctype.item_alternative.item_alternative import get_alternative_items
 from erpnext.stock.doctype.stock_reconciliation.stock_reconciliation import (
 	EmptyStockReconciliationItemsError,
 )
@@ -166,6 +167,192 @@ class TestItemAlternative(ERPNextTestSuite):
 		self.assertEqual(status, True)
 		ste1.submit()
 
+	def test_get_alternative_items_both_directions_and_dedup(self):
+		"""get_alternative_items must return forward alternatives, reverse-only
+		two_way alternatives, exclude one-way reverse rows, and dedupe an item
+		that matches in both the forward and reverse legs of the old UNION."""
+		suffix = random_string(8)
+		base = f"_Test IA Base {suffix}"
+		alt_fwd = f"_Test IA Fwd {suffix}"  # forward only (two_way=0)
+		alt_both = f"_Test IA Both {suffix}"  # forward (two_way=1)
+		alt_rev = f"_Test IA Rev {suffix}"  # reverse via two_way=1
+		alt_norev = f"_Test IA NoRev {suffix}"  # reverse but two_way=0 -> excluded
+		dup = f"_Test IA Dup {suffix}"  # forward AND reverse -> must dedupe
+
+		for item_code in (base, alt_fwd, alt_both, alt_rev, alt_norev, dup):
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		# forward rows: item_code = base
+		make_item_alternative(base, alt_fwd, two_way=0)
+		make_item_alternative(base, alt_both, two_way=1)
+		make_item_alternative(base, dup, two_way=1)
+
+		# reverse rows: alternative_item_code = base
+		make_item_alternative(alt_rev, base, two_way=1)
+		make_item_alternative(alt_norev, base, two_way=0)
+		make_item_alternative(dup, base, two_way=1)
+
+		# txt = the shared suffix so the LIKE matches every alternate but not `base`
+		results = get_alternative_items("Item", suffix, "name", 0, 20, {"item_code": base})
+
+		# structure: list of single-element lists
+		self.assertTrue(all(isinstance(row, list) and len(row) == 1 for row in results))
+
+		returned = [row[0] for row in results]
+
+		# forward alternatives (both one-way and two_way) are returned
+		self.assertIn(alt_fwd, returned)
+		self.assertIn(alt_both, returned)
+
+		# reverse alternative is only returned when the row is two_way
+		self.assertIn(alt_rev, returned)
+		self.assertNotIn(alt_norev, returned)
+
+		# `base` itself is never an alternative of itself
+		self.assertNotIn(base, returned)
+
+		# an item matching both legs of the old UNION is deduped to a single row
+		self.assertIn(dup, returned)
+		self.assertEqual(returned.count(dup), 1)
+
+	def test_get_alternative_items_respects_txt_filter(self):
+		"""The txt LIKE filter must actually narrow the result set so a
+		non-matching alternate is excluded (guards against a broken WHERE)."""
+		suffix = random_string(8)
+		base = f"_Test IA Filter Base {suffix}"
+		matching = f"_Test IA Match {suffix}"
+		other = f"_Test IA Other {suffix}"
+
+		for item_code in (base, matching, other):
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		make_item_alternative(base, matching, two_way=0)
+		make_item_alternative(base, other, two_way=0)
+
+		# search only for the `Match` alternate
+		results = get_alternative_items("Item", f"Match {suffix}", "name", 0, 20, {"item_code": base})
+		returned = [row[0] for row in results]
+
+		self.assertIn(matching, returned)
+		self.assertNotIn(other, returned)
+
+	def test_get_alternative_items_case_insensitive_match(self):
+		"""The txt match must stay case-insensitive on BOTH engines: MariaDB LIKE is
+		case-insensitive by default, and frappe compiles the `like` filter to ILIKE on
+		Postgres. A case-shifted search must still find an alternate whose stored code
+		differs in case — this guards against the conversion degrading to a case-sensitive
+		match (plain LIKE / ==) that would silently return nothing on Postgres."""
+		suffix = random_string(8)
+		base = f"_Test IA Case Base {suffix}"
+		# distinctive mixed-case token in the stored alternate's code
+		alt = f"_Test IA CaseToken AbCdE {suffix}"
+
+		for item_code in (base, alt):
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		make_item_alternative(base, alt, two_way=0)
+
+		# search the LOWERCASED token ("abcde") against the stored "AbCdE"
+		results = get_alternative_items(
+			"Item", f"casetoken abcde {suffix}", "name", 0, 20, {"item_code": base}
+		)
+		returned = [row[0] for row in results]
+
+		self.assertIn(alt, returned)
+
+	def test_get_alternative_items_pagination(self):
+		"""start/page_len must slice the deduped, order-preserving result."""
+		suffix = random_string(8)
+		base = f"_Test IA Page Base {suffix}"
+		alts = [f"_Test IA Page {i} {suffix}" for i in range(3)]
+
+		create_item(base)
+		base_item = frappe.get_doc("Item", base)
+		if not base_item.allow_alternative_item:
+			base_item.allow_alternative_item = 1
+			base_item.save()
+
+		for alt in alts:
+			create_item(alt)
+			alt_item = frappe.get_doc("Item", alt)
+			if not alt_item.allow_alternative_item:
+				alt_item.allow_alternative_item = 1
+				alt_item.save()
+			make_item_alternative(base, alt, two_way=0)
+
+		full = [row[0] for row in get_alternative_items("Item", suffix, "name", 0, 20, {"item_code": base})]
+		self.assertEqual(len(full), 3)
+
+		page = [row[0] for row in get_alternative_items("Item", suffix, "name", 1, 1, {"item_code": base})]
+		self.assertEqual(len(page), 1)
+		self.assertEqual(page[0], full[1])
+
+	def test_get_alternative_items_pagination_is_bounded_and_exact(self):
+		"""Each get_all is bounded to start+page_len rows, so the DB round trip stays small
+		instead of fetching every alternative per keystroke. Walking the result in small pages
+		must still reconstruct the complete deduped set — including an alternate that appears in
+		BOTH legs (forward + reverse two_way) — with no item dropped or duplicated by the bound."""
+		suffix = random_string(8)
+		base = f"_Test IA Bound Base {suffix}"
+		forwards = [f"_Test IA Bound Fwd {i} {suffix}" for i in range(3)]
+		reverses = [f"_Test IA Bound Rev {i} {suffix}" for i in range(3)]
+		dup = f"_Test IA Bound Dup {suffix}"  # forward AND reverse two_way -> deduped across legs
+
+		for item_code in [base, dup, *forwards, *reverses]:
+			create_item(item_code)
+			item = frappe.get_doc("Item", item_code)
+			if not item.allow_alternative_item:
+				item.allow_alternative_item = 1
+				item.save()
+
+		for fwd in forwards:
+			make_item_alternative(base, fwd, two_way=0)
+		make_item_alternative(base, dup, two_way=1)  # dup via the forward leg
+		for rev in reverses:
+			make_item_alternative(rev, base, two_way=1)
+		make_item_alternative(dup, base, two_way=1)  # dup also via the reverse leg
+
+		full = [row[0] for row in get_alternative_items("Item", suffix, "name", 0, 50, {"item_code": base})]
+		# 3 forward + 3 reverse + the single deduped dup = 7 distinct
+		self.assertEqual(len(full), 7)
+		self.assertEqual(full.count(dup), 1)
+
+		# walk in pages of 2; bounded fetches must yield exactly the same set, once each
+		collected = []
+		for start in range(0, 8, 2):
+			collected += [
+				row[0] for row in get_alternative_items("Item", suffix, "name", start, 2, {"item_code": base})
+			]
+
+		self.assertEqual(len(collected), len(set(collected)))  # no duplicates introduced by paging
+		self.assertEqual(set(collected), set(full))  # nothing dropped by the per-leg limit
+		self.assertEqual(collected.count(dup), 1)  # the cross-leg dup survives exactly once
+
+
+def make_item_alternative(item_code, alternative_item_code, two_way=0):
+	doc = frappe.get_doc(
+		{
+			"doctype": "Item Alternative",
+			"item_code": item_code,
+			"alternative_item_code": alternative_item_code,
+			"two_way": two_way,
+		}
+	)
+	doc.insert()
+	return doc
+
 
 def make_items():
 	items = [