fix(manufacturing): keep bom_no/is_phantom_item pair coherent in sub-BOM resolution

When a component is listed more than once in a BOM pointing at different
sub-BOMs (e.g. one phantom, one not), two queries grouped the duplicate
lines into a single row and aggregated bom_no and is_phantom_item with
*independent* Max(). That could pair the phantom flag of one line with the
bom_no of another, so the consumer recursed into the wrong sub-BOM:

- sub_assembly_queries._sub_assembly_rm_query keys on (item_code, bom_no)
  and recurses on is_phantom_item. An incoherent pair sent raw-material
  resolution down the wrong sub-assembly BOM.
- bom_stock_analysis.get_bom_data: explode_phantom_boms recurses into
  bom_no only when is_phantom_item is set; an incoherent pair exploded a
  non-phantom sub-BOM as if it were phantom (or vice-versa).

Fix:
- sub_assembly_queries: group also by (bom_no, is_phantom_item) so each
  distinct sub-BOM is its own coherent row.
- bom_stock_analysis: drop the two independent Max()es and attach a single
  representative line (lowest idx) per item_code before exploding.

This was previously undefined SQL (loose GROUP BY); the fix makes MariaDB
and Postgres agree on a deterministic, coherent pairing. Other Max()-wrapped
columns are functionally dependent on the grouped item and keep their value
on both engines.

Tests (fail on the old code, pass on both engines):
- test_phantom_explosion_picks_coherent_sub_bom: duplicate-component BOM
  explodes the phantom sub-BOM, not the lexically-larger non-phantom one.
- test_sub_assembly_rm_query_keeps_bom_no_phantom_pair_coherent: the query
  returns one coherent row per distinct sub-BOM with the right phantom flag.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Mihir Kandoi
2026-06-18 16:14:30 +05:30
parent eb7f7f2124
commit 41da9eb7fc
4 changed files with 135 additions and 12 deletions

View File

@@ -179,22 +179,25 @@ def _sub_assembly_rm_query(company, bom_no, include_non_stock_items, planned_qty
.on((item.name == item_uom.parent) & (item_uom.uom == item.purchase_uom))
.select(*_sub_assembly_rm_columns(bei, bom, item, item_default, item_uom, planned_qty))
.where(_sub_assembly_rm_filter(bei, bom, item, bom_no, include_non_stock_items))
.groupby(bei.item_code, bei.stock_uom)
.groupby(bei.item_code, bei.stock_uom, bei.bom_no, bei.is_phantom_item)
).run(as_dict=True)
def _sub_assembly_rm_columns(bei, bom, item, item_default, item_uom, planned_qty):
# only item_code/stock_uom are grouped; every other column is functionally dependent on the
# grouped item (item attributes) or arbitrary per BOM Item on MySQL -> Max() keeps the GROUP BY
# valid on postgres while returning the same value MySQL picked.
# Grouped by item_code/stock_uom plus bom_no/is_phantom_item: those two MUST come from the same
# BOM Item row -- the consumer keys on (item_code, bom_no) and recurses on is_phantom_item, so an
# independent Max() per column could pair a bom_no from one line with is_phantom_item from another
# and recurse into the wrong sub-BOM. Grouping them keeps the pair coherent and the GROUP BY valid
# on postgres. The remaining columns are functionally dependent on the grouped item; Max() returns
# their single value on both engines.
return [
(IfNull(Sum(bei.stock_qty / IfNull(bom.quantity, 1)), 0) * planned_qty).as_("qty"),
Max(item.item_name).as_("item_name"),
Max(item.name).as_("item_code"),
Max(bei.description).as_("description"),
bei.stock_uom,
Max(bei.is_phantom_item).as_("is_phantom_item"),
Max(bei.bom_no).as_("bom_no"),
bei.is_phantom_item,
bei.bom_no,
Max(item.min_order_qty).as_("min_order_qty"),
Max(bei.source_warehouse).as_("source_warehouse"),
Max(item.default_material_request_type).as_("default_material_request_type"),

View File

@@ -2862,6 +2862,61 @@ class TestProductionPlan(ERPNextTestSuite):
"The phantom BOM was not re-exploded for the second po_item.",
)
def test_sub_assembly_rm_query_keeps_bom_no_phantom_pair_coherent(self):
"""bom_no and is_phantom_item must stay paired to the same BOM Item line.
When a component is listed more than once in a sub-assembly BOM pointing at different
sub-BOMs (one phantom, one not), grouping only by (item_code, stock_uom) collapsed both
lines into one row, and the independent Max() per column could pair the phantom flag of
one line with the bom_no of the other. The consumer keys on (item_code, bom_no) and
recurses on is_phantom_item, so an incoherent pair recurses into the wrong sub-BOM.
Grouping also by (bom_no, is_phantom_item) yields one coherent row per distinct sub-BOM.
"""
from erpnext.manufacturing.doctype.production_plan.services.sub_assembly_queries import (
_sub_assembly_rm_query,
)
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 first (smaller auto-name); non-phantom second (larger name) -> the name
# the old Max(bom_no) would pick, while Max(is_phantom_item)=1 came from the phantom line.
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])
# Sub-assembly BOM lists `component` twice, once via each sub-BOM.
sa_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
sa_bom = make_bom(item=sa_item, raw_materials=[component], do_not_save=True)
sa_bom.items[0].bom_no = phantom_bom.name
component_doc = frappe.get_doc("Item", component)
sa_bom.append(
"items",
{
"item_code": component,
"qty": 1,
"uom": component_doc.stock_uom,
"stock_uom": component_doc.stock_uom,
"bom_no": normal_bom.name,
},
)
sa_bom.save()
sa_bom.submit()
rows = _sub_assembly_rm_query(
company="_Test Company", bom_no=sa_bom.name, include_non_stock_items=1, planned_qty=1
)
by_bom_no = {row.bom_no: row for row in rows if row.item_code == component}
# One coherent row per distinct sub-BOM, each carrying its own phantom flag.
self.assertIn(phantom_bom.name, by_bom_no)
self.assertIn(normal_bom.name, by_bom_no)
self.assertEqual(by_bom_no[phantom_bom.name].is_phantom_item, 1)
self.assertEqual(by_bom_no[normal_bom.name].is_phantom_item, 0)
def create_production_plan(**args):
"""

View File

@@ -233,13 +233,29 @@ def get_bom_data(filters):
else:
query = query.where(bin.warehouse == filters.get("warehouse"))
if bom_item_table == "BOM Item":
query = query.select(
Max(bom_item.bom_no).as_("bom_no"), Max(bom_item.is_phantom_item).as_("is_phantom_item")
)
data = query.run(as_dict=True)
return explode_phantom_boms(data, filters) if bom_item_table == "BOM Item" else data
if bom_item_table == "BOM Item":
# bom_no + is_phantom_item drive whether/which sub-BOM explode_phantom_boms recurses into, so
# they must come from the SAME BOM Item line. Aggregating each independently (Max) could pair a
# bom_no from one line with is_phantom_item from another when an item_code repeats in the BOM.
# Take the first (lowest idx) line per item_code as the coherent representative.
first_line = {}
for line in frappe.get_all(
"BOM Item",
filters={"parent": filters.get("bom"), "parenttype": "BOM"},
fields=["item_code", "bom_no", "is_phantom_item"],
order_by="idx",
):
first_line.setdefault(line.item_code, line)
for row in data:
line = first_line.get(row.item_code)
if line:
row.bom_no = line.bom_no
row.is_phantom_item = line.is_phantom_item
return explode_phantom_boms(data, filters)
return data
def explode_phantom_boms(data, filters):

View File

@@ -79,6 +79,55 @@ class TestBOMStockAnalysis(ERPNextTestSuite):
)
self.assertEqual(footer.get("description"), expected_min)
def test_phantom_explosion_picks_coherent_sub_bom(self):
"""bom_no and is_phantom_item must come from the SAME BOM Item line.
When a component is listed more than once in a BOM pointing at different sub-BOMs
(one phantom, one not), the report groups both lines into a single row by item_code.
Aggregating bom_no and is_phantom_item with independent Max() could pair the phantom
flag of one line with the bom_no of the other, so explode_phantom_boms recurses into
the wrong sub-BOM. We now take one coherent representative line (lowest idx), so the
phantom sub-BOM is the one exploded.
"""
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; non-phantom second -> larger name,
# which is exactly what the old Max(bom_no) would (incorrectly) 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])
# Parent lists `component` twice: phantom line first (idx 1), non-phantom second (idx 2).
fg_item = make_item(properties={"is_stock_item": 1, "valuation_rate": 10}).name
parent = make_bom(item=fg_item, raw_materials=[component], do_not_save=True)
parent.items[0].bom_no = phantom_bom.name
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": normal_bom.name,
},
)
parent.save()
parent.submit()
raw_data = bom_stock_analysis_report(filters={"qty_to_make": 1, "bom": parent.name})[1]
items = {row.get("item") for row in raw_data if row}
# Phantom sub-BOM exploded -> its raw material appears; the component row is replaced.
self.assertIn(rm_phantom, items)
self.assertNotIn(component, items)
# The non-phantom line's sub-BOM must NOT be mis-exploded.
self.assertNotIn(rm_normal, items)
def split_data_and_footer(raw_data):
"""Separate component rows from the footer row. Skips blank spacer rows."""