From d07967750070a647be42fbde060e261adf6412eb Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Wed, 17 Jun 2026 08:20:23 +0530 Subject: [PATCH 1/2] fix(postgres): db-aware row-locking, savepoints & cursors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PostgreSQL rejects `SELECT ... FOR UPDATE` when combined with `GROUP BY`, aggregates or `DISTINCT`, has no concept of MySQL's locking semantics for those shapes, and its server-side (unbuffered) cursors can't run nested queries mid-iteration. This makes the row-locking / cursor paths db-aware so they keep the exact MariaDB behaviour there and use the valid PostgreSQL form on Postgres. One problem class, applied across the codebase: - **`FOR UPDATE` + GROUP BY/aggregate** — keep `.for_update()` on MariaDB; on Postgres acquire the lock in a separate plain `SELECT ... FOR UPDATE` pass (or skip where the grouped read isn't a lock point). Deprecated serial/batch, serial-batch-bundle, pick list, stock reservation entry. - **Unbuffered/server-side cursor** — Stock Ageing streamed via an unbuffered cursor and then ran nested queries; on Postgres that invalidates the cursor, so process the buffered result directly there. - **Transaction savepoints** — Opening Invoice Creation Tool rolled back the whole transaction per failed invoice (which on Postgres also discards sibling rows and earlier error logs); scope each invoice to a savepoint instead. No behaviour change on MariaDB (the locking/cursor path is unchanged there). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../opening_invoice_creation_tool.py | 9 +++++- erpnext/stock/deprecated_serial_batch.py | 16 ++++++++-- erpnext/stock/doctype/pick_list/pick_list.py | 18 +++++++----- .../stock_reservation_entry.py | 28 +++++++++++------- .../stock/report/stock_ageing/stock_ageing.py | 29 ++++++++++++++----- erpnext/stock/serial_batch_bundle.py | 5 +++- 6 files changed, 75 insertions(+), 30 deletions(-) diff --git a/erpnext/accounts/doctype/opening_invoice_creation_tool/opening_invoice_creation_tool.py b/erpnext/accounts/doctype/opening_invoice_creation_tool/opening_invoice_creation_tool.py index f28c4738c58..a08658b284b 100644 --- a/erpnext/accounts/doctype/opening_invoice_creation_tool/opening_invoice_creation_tool.py +++ b/erpnext/accounts/doctype/opening_invoice_creation_tool/opening_invoice_creation_tool.py @@ -270,6 +270,13 @@ def start_import(invoices): errors = 0 names = [] for idx, d in enumerate(invoices): + # Scope each invoice to a savepoint so a failure only undoes that invoice. + # A plain rollback() would discard the whole transaction — including invoices + # imported earlier in this batch and the error logs of earlier failures (the + # latter only survive on mariadb because the Error Log table is MyISAM; on + # postgres they would be lost). Rolling back to a savepoint keeps both. + savepoint = f"opening_invoice_{frappe.generate_hash(length=8)}" + frappe.db.savepoint(savepoint) try: invoice_number = None if d.invoice_number: @@ -284,7 +291,7 @@ def start_import(invoices): names.append(doc.name) except Exception: errors += 1 - frappe.db.rollback() + frappe.db.rollback(save_point=savepoint) doc.log_error("Opening invoice creation failed") if errors: frappe.msgprint( diff --git a/erpnext/stock/deprecated_serial_batch.py b/erpnext/stock/deprecated_serial_batch.py index 50bcd8416a8..4652f53ff7b 100644 --- a/erpnext/stock/deprecated_serial_batch.py +++ b/erpnext/stock/deprecated_serial_batch.py @@ -146,7 +146,6 @@ class DeprecatedBatchNoValuation: & (sle.batch_no.isnotnull()) & (sle.is_cancelled == 0) ) - .for_update() .groupby(sle.batch_no) ) @@ -156,6 +155,10 @@ class DeprecatedBatchNoValuation: if self.sle.name: query = query.where(sle.name != self.sle.name) + # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres + if frappe.db.db_type != "postgres": + query = query.for_update() + return query.run(as_dict=True) @deprecated( @@ -267,7 +270,6 @@ class DeprecatedBatchNoValuation: & (sle.is_cancelled == 0) & (sle.batch_no.isin(self.non_batchwise_valuation_batches)) ) - .for_update() .where(timestamp_condition) .groupby(sle.batch_no) ) @@ -284,6 +286,10 @@ class DeprecatedBatchNoValuation: query = query.where(batch.use_batchwise_valuation == 0) moving_avg_item_non_batch_value = True + # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres + if frappe.db.db_type != "postgres": + query = query.for_update() + batch_data = query.run(as_dict=True) for d in batch_data: self.available_qty[d.batch_no] += flt(d.batch_qty) @@ -391,7 +397,7 @@ class DeprecatedBatchNoValuation: & (bundle.type_of_transaction.isin(["Inward", "Outward"])) & (bundle_child.batch_no.isin(self.non_batchwise_valuation_batches)) ) - .for_update() + # FOR UPDATE is invalid with GROUP BY on postgres (deprecated valuation path) .where(timestamp_condition) .groupby(bundle_child.batch_no) ) @@ -410,6 +416,10 @@ class DeprecatedBatchNoValuation: query = query.where(batch.use_batchwise_valuation == 0) moving_avg_item_non_batch_value = True + # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres + if frappe.db.db_type != "postgres": + query = query.for_update() + batch_data = query.run(as_dict=True) for d in batch_data: self.available_qty[d.batch_no] += flt(d.batch_qty) diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py index 1baab619740..6ee1b9dede6 100644 --- a/erpnext/stock/doctype/pick_list/pick_list.py +++ b/erpnext/stock/doctype/pick_list/pick_list.py @@ -9,8 +9,7 @@ import frappe from frappe import _, bold from frappe.model.document import Document from frappe.query_builder import Case -from frappe.query_builder.custom import GROUP_CONCAT -from frappe.query_builder.functions import Coalesce, Locate, Replace, Sum +from frappe.query_builder.functions import Coalesce, GroupConcat, Locate, Max, Replace, Sum from frappe.utils import cint, floor, flt, get_link_to_form from frappe.utils.nestedset import get_descendants_of @@ -906,15 +905,16 @@ def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: query = ( frappe.qb.from_(pi_item) .select( - pi_item.sales_order_item, - pi_item.product_bundle_item, - pi_item.item_code, + # only one of sales_order_item / product_bundle_item is grouped per branch below; Max() + # the rest so postgres accepts the query (each is constant within its group) + Max(pi_item.sales_order_item).as_("sales_order_item"), + Max(pi_item.product_bundle_item).as_("product_bundle_item"), + Max(pi_item.item_code).as_("item_code"), pi_item.sales_order, Sum(pi_item.stock_qty).as_("stock_qty"), Sum(pi_item.picked_qty).as_("picked_qty"), ) .where(pi_item.docstatus == 1) - .for_update() ) if contains_packed_items: @@ -928,6 +928,10 @@ def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: pi_item.sales_order, ).where(pi_item.sales_order_item.isin(items)) + # FOR UPDATE is invalid with GROUP BY on postgres; lock scanned rows on MariaDB only + if frappe.db.db_type != "postgres": + query = query.for_update() + return query.run(as_dict=True) @@ -1365,7 +1369,7 @@ def get_pick_list_query(doctype: Any, txt: str, searchfield: Any, start: int, pa .select( PICK_LIST.name, SALES_ORDER.customer, - Replace(GROUP_CONCAT(PICK_LIST_ITEM.sales_order).distinct(), ",", "
").as_("sales_order"), + Replace(GroupConcat(PICK_LIST_ITEM.sales_order).distinct(), ",", "
").as_("sales_order"), ) .where(PICK_LIST.docstatus == 1) .where(PICK_LIST.status.isin(["Open", "Partly Delivered"])) diff --git a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py index b19f8db2052..2c5c89596ad 100644 --- a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py +++ b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py @@ -8,7 +8,7 @@ import frappe from frappe import _ from frappe.model.document import Document from frappe.query_builder import Case -from frappe.query_builder.functions import Sum +from frappe.query_builder.functions import Max, Min, Sum from frappe.utils import cint, flt, nowdate, nowtime, parse_json from erpnext.stock.utils import get_or_make_bin, get_stock_balance @@ -715,12 +715,15 @@ def get_available_qty_to_reserve( & (sre.warehouse == warehouse) & (sre.delivered_qty < sre.reserved_qty) ) - .for_update() ) if ignore_sre: query = query.where(sre.name != ignore_sre) + # FOR UPDATE is invalid with aggregates on postgres; lock scanned rows on MariaDB only + if frappe.db.db_type != "postgres": + query = query.for_update() + reserved_qty = query.run()[0][0] or 0.0 if reserved_qty: @@ -870,14 +873,16 @@ def get_sre_reserved_warehouses_for_voucher( query = ( frappe.qb.from_(sre) .select(sre.warehouse) - .distinct() .where( (sre.docstatus == 1) & (sre.voucher_type == voucher_type) & (sre.voucher_no == voucher_no) & (sre.delivered_qty < sre.reserved_qty) ) - .orderby(sre.creation) + # distinct warehouses, earliest reservation first (postgres can't ORDER BY a + # non-selected column under SELECT DISTINCT, so group + Min instead) + .groupby(sre.warehouse) + .orderby(Min(sre.creation)) ) if voucher_detail_no: @@ -984,7 +989,8 @@ def get_sre_reserved_batch_nos_details(item_code: str, warehouse: str, batch_nos & (sre.reservation_based_on == "Serial and Batch") ) .groupby(sb_entry.batch_no) - .orderby(sb_entry.creation) + # result is collapsed into a dict below, so ordering is irrelevant; dropping the (non-grouped) + # ORDER BY creation keeps the GROUP BY valid on postgres. ) if batch_nos: @@ -1526,10 +1532,12 @@ class StockReservation: .inner_join(child_doctype) .on(doctype.name == child_doctype.parent) .select( - doctype.name.as_("voucher_no"), + # grouped by the child PK (name), so child columns are valid on postgres via functional + # dependency; the parent (doctype) columns aren't, so Max() them -- constant per child row. + Max(doctype.name).as_("voucher_no"), child_doctype.name.as_("voucher_detail_no"), child_doctype[item_code_fieldname].as_("item_code"), - doctype.company, + Max(doctype.company).as_("company"), child_doctype.stock_uom, ) .where((doctype.docstatus == 1) & (doctype[field].isin(docnames))) @@ -1539,9 +1547,9 @@ class StockReservation: if to_doctype == "Work Order": query = query.select( child_doctype.source_warehouse, - doctype.wip_warehouse, - doctype.skip_transfer, - doctype.from_wip_warehouse, + Max(doctype.wip_warehouse).as_("wip_warehouse"), + Max(doctype.skip_transfer).as_("skip_transfer"), + Max(doctype.from_wip_warehouse).as_("from_wip_warehouse"), child_doctype.required_qty, (child_doctype.required_qty - child_doctype.transferred_qty).as_("qty"), child_doctype.stock_reserved_qty, diff --git a/erpnext/stock/report/stock_ageing/stock_ageing.py b/erpnext/stock/report/stock_ageing/stock_ageing.py index e1fb482ab63..9d1acf9b243 100644 --- a/erpnext/stock/report/stock_ageing/stock_ageing.py +++ b/erpnext/stock/report/stock_ageing/stock_ageing.py @@ -309,20 +309,31 @@ class FIFOSlots: self.prepare_stock_reco_voucher_wise_count() if stock_ledger_entries is None: - # nested queries invalidate the streaming cursor below, + # streaming path: nested queries invalidate the streaming cursor below, # so batchwise valuation flags must be resolved beforehand self._prefetch_batchwise_valuations() - with frappe.db.unbuffered_cursor(): - if stock_ledger_entries is None: - stock_ledger_entries = self._get_stock_ledger_entries() + if frappe.db.db_type == "postgres": + # postgres server-side cursors can't run nested queries mid-iteration; _get_stock_ledger_entries + # returns a buffered result there, so process it directly (no unbuffered cursor). + for row in self._get_stock_ledger_entries(): + self._process_stock_ledger_entry(row, bundle_wise_serial_nos, bundle_wise_batch_nos) + else: + with frappe.db.unbuffered_cursor(): + stock_ledger_entries = self._get_stock_ledger_entries() + for row in stock_ledger_entries: + self._process_stock_ledger_entry(row, bundle_wise_serial_nos, bundle_wise_batch_nos) + + # Note that stock_ledger_entries is an iterator, you can not reuse it like a list + del stock_ledger_entries + else: + # entries passed in directly as a list: no streaming cursor is opened, so the batchwise + # valuation flags can be resolved lazily — a nested get_value here is safe on postgres too + # (running it inside an unbuffered/named cursor would raise on postgres). for row in stock_ledger_entries: self._process_stock_ledger_entry(row, bundle_wise_serial_nos, bundle_wise_batch_nos) - # Note that stock_ledger_entries is an iterator, you can not reuse it like a list - del stock_ledger_entries - if not self.filters.get("show_warehouse_wise_stock"): # (Item 1, WH 1), (Item 1, WH 2) => (Item 1) self.item_details = self._aggregate_details_by_item(self.item_details) @@ -944,7 +955,9 @@ class FIFOSlots: sle_query = sle_query.orderby(sle.posting_datetime, sle.creation) - return sle_query.run(as_dict=True, as_iterator=True) + # postgres server-side (named) cursors can't run nested queries mid-iteration, which + # _process_stock_ledger_entry needs; fall back to a buffered fetch there. MariaDB streams. + return sle_query.run(as_dict=True, as_iterator=frappe.db.db_type != "postgres") def _get_bundle_wise_serial_nos(self) -> dict: bundle = frappe.qb.DocType("Serial and Batch Bundle") diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py index 4949e389b48..2e9a9a0d97b 100644 --- a/erpnext/stock/serial_batch_bundle.py +++ b/erpnext/stock/serial_batch_bundle.py @@ -865,10 +865,13 @@ class BatchNoValuation(DeprecatedBatchNoValuation): & (child.docstatus == 1) & (child.type_of_transaction.isin(["Inward", "Outward"])) ) - .for_update() .groupby(child.batch_no) ) + # FOR UPDATE is invalid with GROUP BY on postgres; lock scanned rows on MariaDB only + if frappe.db.db_type != "postgres": + query = query.for_update() + # Important to exclude the current voucher detail no / voucher no to calculate the correct stock value difference if self.sle.voucher_detail_no: query = query.where(child.voucher_detail_no != self.sle.voucher_detail_no) From a8991830877c35d9668aa3d7cd68e3b181023584 Mon Sep 17 00:00:00 2001 From: Mihir Kandoi Date: Wed, 17 Jun 2026 08:50:33 +0530 Subject: [PATCH 2/2] fix(postgres): keep row locks via lock-then-read (address review) Acquire the same row locks on postgres that MariaDB takes, via a separate plain SELECT ... FOR UPDATE before each grouped/aggregate read (FOR UPDATE is invalid with GROUP BY on postgres). Applied to all 6 aggregate lock sites. Co-Authored-By: Claude Opus 4.8 (1M context) --- erpnext/stock/deprecated_serial_batch.py | 112 +++++++++++------- erpnext/stock/doctype/pick_list/pick_list.py | 23 ++-- .../stock_reservation_entry.py | 27 +++-- erpnext/stock/serial_batch_bundle.py | 44 ++++--- 4 files changed, 119 insertions(+), 87 deletions(-) diff --git a/erpnext/stock/deprecated_serial_batch.py b/erpnext/stock/deprecated_serial_batch.py index 4652f53ff7b..b2010411644 100644 --- a/erpnext/stock/deprecated_serial_batch.py +++ b/erpnext/stock/deprecated_serial_batch.py @@ -132,6 +132,24 @@ class DeprecatedBatchNoValuation: sle.creation < self.sle.creation ) + conditions = ( + (sle.item_code == self.sle.item_code) + & (sle.warehouse == self.sle.warehouse) + & (sle.batch_no.isin(self.batchwise_valuation_batches)) + & (sle.batch_no.isnotnull()) + & (sle.is_cancelled == 0) + ) + if timestamp_condition: + conditions &= timestamp_condition + if self.sle.name: + conditions &= sle.name != self.sle.name + + # Lock the scanned SLE rows so a concurrent stock posting can't change them mid-valuation. + # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so + # lock the same rows in a separate plain SELECT first (held for the transaction). + if frappe.db.db_type == "postgres": + frappe.qb.from_(sle).select(sle.name).where(conditions).for_update().run() + query = ( frappe.qb.from_(sle) .select( @@ -139,23 +157,9 @@ class DeprecatedBatchNoValuation: Sum(sle.stock_value_difference).as_("batch_value"), Sum(sle.actual_qty).as_("batch_qty"), ) - .where( - (sle.item_code == self.sle.item_code) - & (sle.warehouse == self.sle.warehouse) - & (sle.batch_no.isin(self.batchwise_valuation_batches)) - & (sle.batch_no.isnotnull()) - & (sle.is_cancelled == 0) - ) + .where(conditions) .groupby(sle.batch_no) ) - - if timestamp_condition: - query = query.where(timestamp_condition) - - if self.sle.name: - query = query.where(sle.name != self.sle.name) - - # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres if frappe.db.db_type != "postgres": query = query.for_update() @@ -254,6 +258,24 @@ class DeprecatedBatchNoValuation: sle.creation < self.sle.creation ) + conditions = ( + (sle.item_code == self.sle.item_code) + & (sle.warehouse == self.sle.warehouse) + & (sle.batch_no.isnotnull()) + & (sle.is_cancelled == 0) + & (sle.batch_no.isin(self.non_batchwise_valuation_batches)) + & timestamp_condition + ) + if self.sle.name: + conditions &= sle.name != self.sle.name + + # Lock the scanned SLE rows so a concurrent stock posting can't change them mid-valuation. + # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so + # lock the same SLE rows in a separate plain SELECT first. The batch.use_batchwise_valuation + # refinement below only narrows the set, so locking without the join is a safe superset. + if frappe.db.db_type == "postgres": + frappe.qb.from_(sle).select(sle.name).where(conditions).for_update().run() + query = ( frappe.qb.from_(sle) .inner_join(batch) @@ -263,20 +285,10 @@ class DeprecatedBatchNoValuation: Sum(sle.actual_qty).as_("batch_qty"), Sum(sle.stock_value_difference).as_("batch_value"), ) - .where( - (sle.item_code == self.sle.item_code) - & (sle.warehouse == self.sle.warehouse) - & (sle.batch_no.isnotnull()) - & (sle.is_cancelled == 0) - & (sle.batch_no.isin(self.non_batchwise_valuation_batches)) - ) - .where(timestamp_condition) + .where(conditions) .groupby(sle.batch_no) ) - if self.sle.name: - query = query.where(sle.name != self.sle.name) - # Moving Average items with no Use Batch wise Valuation but want to use batch wise valuation moving_avg_item_non_batch_value = False if valuation_method := self.get_valuation_method(self.sle.item_code): @@ -286,7 +298,6 @@ class DeprecatedBatchNoValuation: query = query.where(batch.use_batchwise_valuation == 0) moving_avg_item_non_batch_value = True - # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres if frappe.db.db_type != "postgres": query = query.for_update() @@ -377,6 +388,35 @@ class DeprecatedBatchNoValuation: bundle.creation < self.sle.creation ) + conditions = ( + (bundle.item_code == self.sle.item_code) + & (bundle.warehouse == self.sle.warehouse) + & (bundle_child.batch_no.isnotnull()) + & (bundle.is_cancelled == 0) + & (bundle.docstatus == 1) + & (bundle.type_of_transaction.isin(["Inward", "Outward"])) + & (bundle_child.batch_no.isin(self.non_batchwise_valuation_batches)) + & timestamp_condition + ) + if self.sle.serial_and_batch_bundle: + conditions &= bundle.name != self.sle.serial_and_batch_bundle + conditions &= bundle.voucher_type != "Pick List" + + # Lock the scanned bundle rows so a concurrent stock posting can't change them mid-valuation. + # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so + # lock the same rows in a separate plain SELECT first (the batch.use_batchwise_valuation + # refinement below only narrows the set, so omitting that join is a safe superset). + if frappe.db.db_type == "postgres": + ( + frappe.qb.from_(bundle) + .inner_join(bundle_child) + .on(bundle.name == bundle_child.parent) + .select(bundle_child.name) + .where(conditions) + .for_update() + .run() + ) + query = ( frappe.qb.from_(bundle) .inner_join(bundle_child) @@ -388,25 +428,10 @@ class DeprecatedBatchNoValuation: Sum(bundle_child.qty).as_("batch_qty"), Sum(bundle_child.stock_value_difference).as_("batch_value"), ) - .where( - (bundle.item_code == self.sle.item_code) - & (bundle.warehouse == self.sle.warehouse) - & (bundle_child.batch_no.isnotnull()) - & (bundle.is_cancelled == 0) - & (bundle.docstatus == 1) - & (bundle.type_of_transaction.isin(["Inward", "Outward"])) - & (bundle_child.batch_no.isin(self.non_batchwise_valuation_batches)) - ) - # FOR UPDATE is invalid with GROUP BY on postgres (deprecated valuation path) - .where(timestamp_condition) + .where(conditions) .groupby(bundle_child.batch_no) ) - if self.sle.serial_and_batch_bundle: - query = query.where(bundle.name != self.sle.serial_and_batch_bundle) - - query = query.where(bundle.voucher_type != "Pick List") - # Moving Average items with no Use Batch wise Valuation but want to use batch wise valuation moving_avg_item_non_batch_value = False if valuation_method := self.get_valuation_method(self.sle.item_code): @@ -416,7 +441,6 @@ class DeprecatedBatchNoValuation: query = query.where(batch.use_batchwise_valuation == 0) moving_avg_item_non_batch_value = True - # lock scanned rows on MariaDB; FOR UPDATE is invalid with GROUP BY on postgres if frappe.db.db_type != "postgres": query = query.for_update() diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py index 6ee1b9dede6..2429970dd50 100644 --- a/erpnext/stock/doctype/pick_list/pick_list.py +++ b/erpnext/stock/doctype/pick_list/pick_list.py @@ -902,6 +902,9 @@ def update_pick_list_status(pick_list): def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: pi_item = frappe.qb.DocType("Pick List Item") + group_field = pi_item.product_bundle_item if contains_packed_items else pi_item.sales_order_item + conditions = (pi_item.docstatus == 1) & group_field.isin(items) + query = ( frappe.qb.from_(pi_item) .select( @@ -914,22 +917,16 @@ def get_picked_items_qty(items, contains_packed_items=False) -> list[dict]: Sum(pi_item.stock_qty).as_("stock_qty"), Sum(pi_item.picked_qty).as_("picked_qty"), ) - .where(pi_item.docstatus == 1) + .where(conditions) + .groupby(group_field, pi_item.sales_order) ) - if contains_packed_items: - query = query.groupby( - pi_item.product_bundle_item, - pi_item.sales_order, - ).where(pi_item.product_bundle_item.isin(items)) + # Lock the picked-qty rows so a concurrent pick can't change them mid-transaction. MariaDB carries + # the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so lock the same rows + # in a separate plain SELECT first (held for the transaction). + if frappe.db.db_type == "postgres": + frappe.qb.from_(pi_item).select(pi_item.name).where(conditions).for_update().run() else: - query = query.groupby( - pi_item.sales_order_item, - pi_item.sales_order, - ).where(pi_item.sales_order_item.isin(items)) - - # FOR UPDATE is invalid with GROUP BY on postgres; lock scanned rows on MariaDB only - if frappe.db.db_type != "postgres": query = query.for_update() return query.run(as_dict=True) diff --git a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py index 2c5c89596ad..5c586a1fd53 100644 --- a/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py +++ b/erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py @@ -706,21 +706,26 @@ def get_available_qty_to_reserve( if available_qty: sre = frappe.qb.DocType("Stock Reservation Entry") + conditions = ( + (sre.docstatus == 1) + & (sre.item_code == item_code) + & (sre.warehouse == warehouse) + & (sre.delivered_qty < sre.reserved_qty) + ) + if ignore_sre: + conditions &= sre.name != ignore_sre + + # Lock the rows being aggregated so a concurrent reservation can't change them mid-transaction. + # MariaDB carries the lock on the aggregate query itself; postgres rejects FOR UPDATE with an + # aggregate, so on postgres lock the same rows in a separate plain SELECT first (held for the txn). + if frappe.db.db_type == "postgres": + frappe.qb.from_(sre).select(sre.name).where(conditions).for_update().run() + query = ( frappe.qb.from_(sre) .select(Sum(sre.reserved_qty - sre.delivered_qty - sre.transferred_qty - sre.consumed_qty)) - .where( - (sre.docstatus == 1) - & (sre.item_code == item_code) - & (sre.warehouse == warehouse) - & (sre.delivered_qty < sre.reserved_qty) - ) + .where(conditions) ) - - if ignore_sre: - query = query.where(sre.name != ignore_sre) - - # FOR UPDATE is invalid with aggregates on postgres; lock scanned rows on MariaDB only if frappe.db.db_type != "postgres": query = query.for_update() diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py index 2e9a9a0d97b..1144f32f848 100644 --- a/erpnext/stock/serial_batch_bundle.py +++ b/erpnext/stock/serial_batch_bundle.py @@ -851,6 +851,30 @@ class BatchNoValuation(DeprecatedBatchNoValuation): child.creation < self.sle.creation ) + conditions = ( + (child.item_code == self.sle.item_code) + & (child.warehouse == self.sle.warehouse) + & (child.batch_no.isin(self.batchwise_valuation_batches)) + & (child.docstatus == 1) + & (child.type_of_transaction.isin(["Inward", "Outward"])) + ) + + # Important to exclude the current voucher detail no / voucher no to calculate the correct stock value difference + if self.sle.voucher_detail_no: + conditions &= child.voucher_detail_no != self.sle.voucher_detail_no + elif self.sle.voucher_no: + conditions &= child.voucher_no != self.sle.voucher_no + + conditions &= child.voucher_type != "Pick List" + if timestamp_condition: + conditions &= timestamp_condition + + # Lock the scanned rows so a concurrent stock transaction can't change them mid-valuation. + # MariaDB carries the lock on the grouped query; postgres rejects FOR UPDATE with GROUP BY, so + # lock the same rows in a separate plain SELECT first (held for the transaction). + if frappe.db.db_type == "postgres": + frappe.qb.from_(child).select(child.name).where(conditions).for_update().run() + query = ( frappe.qb.from_(child) .select( @@ -858,30 +882,12 @@ class BatchNoValuation(DeprecatedBatchNoValuation): Sum(child.stock_value_difference).as_("incoming_rate"), Sum(child.qty).as_("qty"), ) - .where( - (child.item_code == self.sle.item_code) - & (child.warehouse == self.sle.warehouse) - & (child.batch_no.isin(self.batchwise_valuation_batches)) - & (child.docstatus == 1) - & (child.type_of_transaction.isin(["Inward", "Outward"])) - ) + .where(conditions) .groupby(child.batch_no) ) - - # FOR UPDATE is invalid with GROUP BY on postgres; lock scanned rows on MariaDB only if frappe.db.db_type != "postgres": query = query.for_update() - # Important to exclude the current voucher detail no / voucher no to calculate the correct stock value difference - if self.sle.voucher_detail_no: - query = query.where(child.voucher_detail_no != self.sle.voucher_detail_no) - elif self.sle.voucher_no: - query = query.where(child.voucher_no != self.sle.voucher_no) - - query = query.where(child.voucher_type != "Pick List") - if timestamp_condition: - query = query.where(timestamp_condition) - return query.run(as_dict=True) def prepare_batches(self):