Merge pull request #56893 from mihir-kandoi/pg-bom-groupby-phantom-pair

fix(manufacturing): keep bom_no/is_phantom_item pair coherent in get_bom_items_as_dict

erpnext/manufacturing/doctype/bom/bom.py

@@ -1319,8 +1319,12 @@ def _add_secondary_item_columns(query, t, stock_item_condition):
 
 
 def _add_normal_item_columns(query, t, amount_col, stock_item_condition, track_semi_finished_goods):
-	# non-grouped columns are constant per grouped item_code (+operation/operation_row_id) -> Max()
+	# Grouped also by bom_no/is_phantom_item: the pair MUST come from the same BOM Item row --
-	# keeps the GROUP BY valid on postgres while returning the value MySQL picked arbitrarily.
+	# _add_bom_item_to_dict recurses into bom_no when is_phantom_item is set, so independent Max()
+	# per column could pair one line's phantom flag with another line's bom_no and explode the
+	# wrong sub-BOM (same fix as sub_assembly_queries). The remaining non-grouped columns are
+	# constant per grouped item_code (+operation/operation_row_id) -> Max() keeps the GROUP BY
+	# valid on postgres while returning the value MySQL picked arbitrarily.
 	# NOTE: base_rate is aliased "rate" below and is what callers receive; bom_item.rate was selected
 	# under the same alias and silently shadowed (last value wins in the dict), so it is dropped here
 	# -- output is unchanged.
@@ -1335,14 +1339,15 @@ def _add_normal_item_columns(query, t, amount_col, stock_item_condition, track_s
 		Max(t.bom_item.description).as_("description"),
 		Max(t.bom_item.base_rate).as_("rate"),
 		Max(t.bom_item.operation_row_id).as_("operation_row_id"),
-		Max(t.bom_item.is_phantom_item).as_("is_phantom_item"),
+		t.bom_item.is_phantom_item,
-		Max(t.bom_item.bom_no).as_("bom_no"),
+		t.bom_item.bom_no,
 	).where(stock_item_condition | (t.bom_item.is_phantom_item == 1))
 
 	if track_semi_finished_goods:
 		group_by = [t.bom_item.item_code, t.bom_item.operation_row_id, t.item_doc.stock_uom]
 	else:
 		group_by = [t.bom_item.item_code, t.item_doc.stock_uom, t.bom_item.operation]
+	group_by += [t.bom_item.bom_no, t.bom_item.is_phantom_item]
 
 	return query, group_by
 

erpnext/manufacturing/doctype/bom/test_bom.py

@@ -57,6 +57,50 @@ class TestBOM(ERPNextTestSuite):
 
 		self.assertEqual(len(get_bom_items(bom=get_default_bom(), company="_Test Company")), 3)
 
+	@timeout
+	def test_get_items_keeps_bom_no_phantom_pair_coherent(self):
+		from erpnext.manufacturing.doctype.bom.bom import get_bom_items_as_dict
+		from erpnext.manufacturing.doctype.production_plan.test_production_plan import make_bom
+
+		for phantom_first in (True, False):
+			rm_phantom = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+			rm_normal = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+			component = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+
+			# phantom sub-BOM created first -> smaller auto-name; the non-phantom one gets the
+			# larger name, which is exactly what an independent Max(bom_no) would wrongly pick
+			phantom_bom = make_bom(item=component, raw_materials=[rm_phantom], do_not_save=True)
+			phantom_bom.is_phantom_bom = 1
+			phantom_bom.save()
+			phantom_bom.submit()
+			normal_bom = make_bom(item=component, raw_materials=[rm_normal])
+
+			fg_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
+			first_bom, second_bom = (
+				(phantom_bom.name, normal_bom.name) if phantom_first else (normal_bom.name, phantom_bom.name)
+			)
+			parent = make_bom(item=fg_item, raw_materials=[component], do_not_save=True)
+			parent.items[0].bom_no = first_bom
+			component_doc = frappe.get_doc("Item", component)
+			parent.append(
+				"items",
+				{
+					"item_code": component,
+					"qty": 1,
+					"uom": component_doc.stock_uom,
+					"stock_uom": component_doc.stock_uom,
+					"bom_no": second_bom,
+				},
+			)
+			parent.save()
+			parent.submit()
+
+			items_dict = get_bom_items_as_dict(parent.name, "_Test Company", qty=1, fetch_exploded=0)
+			self.assertIn(rm_phantom, items_dict)
+			self.assertIn(component, items_dict)
+			self.assertEqual(flt(items_dict[component].qty), 1.0)
+			self.assertNotIn(rm_normal, items_dict)
+
 	@timeout
 	def test_default_bom(self):
 		def _get_default_bom_in_item():