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..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,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)
diff --git a/erpnext/stock/doctype/pick_list/pick_list.py b/erpnext/stock/doctype/pick_list/pick_list.py
index 1baab619740..2429970dd50 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
@@ -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(), ",", "
").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..5c586a1fd53 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
@@ -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,
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..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,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)