nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-08-12 14:11:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #56025 from mihir-kandoi/pg-row-locking-cursor

Browse Source 
fix(postgres): db-aware row-locking, savepoints & cursors
This commit is contained in:
Mihir Kandoi
2026-06-17 12:19:55 +05:30
committed by GitHub
parent faadc1620b a899183087
commit 48e66d04e6
6 changed files with 182 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
erpnext/accounts/doctype/opening_invoice_creation_tool/opening_invoice_creation_tool.py
View File

@@ -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(

118
erpnext/stock/deprecated_serial_batch.py
View File

@@ -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,22 +157,11 @@ 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)
)
.for_update()
.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)
if frappe.db.db_type != "postgres":
query = query.for_update()
return query.run(as_dict=True)
@@ -251,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)
@@ -260,21 +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))
)
.for_update()
.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):
@@ -284,6 +298,9 @@ class DeprecatedBatchNoValuation:
query = query.where(batch.use_batchwise_valuation == 0)
moving_avg_item_non_batch_value = True
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)
@@ -371,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)
@@ -382,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()
.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):
@@ -410,6 +441,9 @@ class DeprecatedBatchNoValuation:
query = query.where(batch.use_batchwise_valuation == 0)
moving_avg_item_non_batch_value = True
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)

35
erpnext/stock/doctype/pick_list/pick_list.py
View File

@@ -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
@@ -903,30 +902,32 @@ 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(
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()
.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))
query = query.for_update()
return query.run(as_dict=True)
@@ -1365,7 +1366,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(), ",", "<br>").as_("sales_order"),
Replace(GroupConcat(PICK_LIST_ITEM.sales_order).distinct(), ",", "<br>").as_("sales_order"),
)
.where(PICK_LIST.docstatus == 1)
.where(PICK_LIST.status.isin(["Open", "Partly Delivered"]))

51
erpnext/stock/doctype/stock_reservation_entry/stock_reservation_entry.py
View File

@@ -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
@@ -706,20 +706,28 @@ 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)
)
.for_update()
.where(conditions)
)
if ignore_sre:
query = query.where(sre.name != ignore_sre)
if frappe.db.db_type != "postgres":
query = query.for_update()
reserved_qty = query.run()[0][0] or 0.0
@@ -870,14 +878,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 +994,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 +1537,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 +1552,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,

29
erpnext/stock/report/stock_ageing/stock_ageing.py
View File

@@ -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")

45
erpnext/stock/serial_batch_bundle.py
View File

@@ -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,26 +882,11 @@ 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"]))
)
.for_update()
.where(conditions)
.groupby(child.batch_no)
)
# 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)
if frappe.db.db_type != "postgres":
query = query.for_update()
return query.run(as_dict=True)