From 3a8bd852d5c91987c6a86d9756a48e7d26b82cc8 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Fri, 31 Jul 2026 10:14:50 +0530 Subject: [PATCH 1/5] refactor: rebuild pcv on mapreduce (parallelization) --- .../period_closing_voucher.py | 180 +++++++++++++++++- 1 file changed, 175 insertions(+), 5 deletions(-) diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py index 1f4a60e6f14..9d424f9b893 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py @@ -3,11 +3,22 @@ import copy +from datetime import timedelta import frappe -from frappe import _ -from frappe.query_builder.functions import Max, Sum -from frappe.utils import add_days, flt, fmt_money, formatdate, get_link_to_form, getdate +from frappe import _, qb +from frappe.query_builder.functions import Max, Min, Sum +from frappe.utils import ( + add_days, + ceil, + cint, + flt, + fmt_money, + formatdate, + get_datetime, + get_link_to_form, + getdate, +) from erpnext import is_perpetual_inventory_enabled from erpnext.accounts.doctype.account_closing_balance.account_closing_balance import ( @@ -265,8 +276,14 @@ class PeriodClosingVoucher(AccountsController): if frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): self.make_gl_entries() else: - ppcv = frappe.get_doc({"doctype": "Process Period Closing Voucher", "parent_pcv": self.name}) - ppcv.save().submit() + from frappe.utils.background_jobs import mapreduce + + data = self.get_data_for_mapreduce() + mapreduce( + "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.mapper", + "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.reducer", + data, + ) def on_cancel(self): self.ignore_linked_doctypes = ( @@ -594,6 +611,91 @@ class PeriodClosingVoucher(AccountsController): {"voucher_type": "Period Closing Voucher", "voucher_no": self.name, "is_cancelled": 0}, ) + def get_data_for_mapreduce(self): + return self.generate_tasks_for_normal_balance() + self.generate_tasks_for_opening_balance() + + def get_period_range_for_tasks(self, start_date, end_date, step_size, report_type, balance_type): + start_date = getdate(start_date) + end_date = getdate(end_date) + + # split period into date ranges + curr_date = getdate(start_date) + date_splits = [] + while True: + next_date = getdate(add_days(curr_date, step_size)) + if next_date < end_date: + date_splits.append( + { + "from_date": str(curr_date), + "to_date": str(next_date), + "pcv": self.name, + "report_type": report_type, + "balance_type": balance_type, + } + ) + curr_date = getdate(add_days(next_date, 1)) + else: + date_splits.append( + { + "from_date": str(curr_date), + "to_date": str(end_date), + "pcv": self.name, + "report_type": report_type, + "balance_type": balance_type, + } + ) + break + + return date_splits + + def generate_tasks_for_normal_balance(self): + # estimation can be wrong by a factor of 2 + estimated_count = ( + cint( + frappe.db.sql( + f"explain select count(*) from `tabGL Entry` where is_cancelled = 0 and posting_date between {self.period_start_date} and {self.period_end_date};", + as_dict=True, + )[0].rows + ) + * 2 + ) + job_count = ( + 1 if estimated_count / 2000000 else ceil(estimated_count / 2000000) + ) # conservative chunk size + days = (getdate(self.period_end_date) - getdate(self.period_start_date)).days + step_size = 1 if days / job_count < 1 else ceil(days / job_count) + return self.get_period_range_for_tasks( + self.period_start_date, self.period_end_date, step_size, "Balance Sheet", "Normal Balance" + ) + self.get_period_range_for_tasks( + self.period_start_date, self.period_end_date, step_size, "Profit and Loss", "Normal Balance" + ) + + def generate_tasks_for_opening_balance(self): + tasks = [] + if self.is_first_period_closing_voucher(): + gl = qb.DocType("GL Entry") + min = qb.from_(gl).select(Min(gl.posting_date)).run()[0][0] + max = qb.from_(gl).select(Max(gl.posting_date)).run()[0][0] + + # estimation can be wrong by a factor of 2 + estimated_count = ( + cint( + frappe.db.sql( + f"explain select count(*) from `tabGL Entry` where is_cancelled = 0 and is_opening = 0 and posting_date between {min} and {max};", + as_dict=True, + )[0].rows + ) + * 2 + ) + job_count = ( + 1 if estimated_count / 2000000 else ceil(estimated_count / 2000000) + ) # conservative chunk size + days = (getdate(self.period_end_date) - getdate(self.period_start_date)).days + step_size = 1 if days / job_count < 1 else ceil(days / job_count) + tasks = self.get_period_range_for_tasks(min, max, step_size, "Balance Sheet", "Opening Balance") + + return tasks + def process_gl_and_closing_entries(doc): from erpnext.accounts.general_ledger import make_gl_entries @@ -673,3 +775,71 @@ def get_previous_closed_period_in_current_year(fiscal_year, company): order_by="period_end_date desc", ) return prev_closed_period_end_date + + +def mapper(val): + start_date = val.from_date + end_date = val.to_date + pcv = val.pcv + report_type = val.report_type + balance_type = val.balance_type + company = frappe.db.get_value("Period Closing Voucher", pcv, "company") + dimensions = get_dimensions() + + accounts = frappe.db.get_all( + "Account", filters={"company": company, "report_type": report_type}, pluck="name" + ) + + # summarize + gle = qb.DocType("GL Entry") + query = qb.from_(gle).select(gle.account) + for dim in dimensions: + query = query.select(gle[dim]) + query = query.select( + Sum(gle.debit).as_("debit"), + Sum(gle.credit).as_("credit"), + Sum(gle.debit_in_account_currency).as_("debit_in_account_currency"), + Sum(gle.credit_in_account_currency).as_("credit_in_account_currency"), + # account_currency is constant per grouped account -> Max() keeps the GROUP BY postgres-valid + Max(gle.account_currency).as_("account_currency"), + ).where( + (gle.company.eq(company)) + & (gle.is_cancelled.eq(0)) + & (gle.posting_date.between(start_date, end_date)) + & (gle.account.isin(accounts)) + ) + + if balance_type == "Opening Balance": + query = query.where(gle.is_opening.eq("Yes")) + else: + # Keep balances aligned with legacy PCV logic (non-opening transactions only) + query = query.where(gle.is_opening.eq("No")) + + query = query.groupby(gle.account) + for dim in dimensions: + query = query.groupby(gle[dim]) + + res = query.run(as_dict=True) + return res + + +def reducer(final, partial_res): + if final is None: + final = [] + + gl_entries = [] + if partial_res: + for x in partial_res: + gl_entries.append(frappe._dict(x)) + + return final + gl_entries + + +def get_dimensions(): + from erpnext.accounts.doctype.accounting_dimension.accounting_dimension import ( + get_accounting_dimensions, + ) + + default_dimensions = ["cost_center", "finance_book", "project"] + dimensions = default_dimensions + get_accounting_dimensions() + return dimensions From 503a80f2c998d3d0c521b7897506c155ca9eb878 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Mon, 31 Aug 2026 16:37:50 +0530 Subject: [PATCH 2/5] refactor: dynamic link to mapreduce and clean on cancel and trash --- .../period_closing_voucher.js | 2 +- .../period_closing_voucher.py | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js index 27a38912a86..edfde8f52e1 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.js @@ -5,7 +5,7 @@ frappe.ui.form.on("Period Closing Voucher", { onload: function (frm) { if (!frm.doc.transaction_date) frm.doc.transaction_date = frappe.datetime.obj_to_str(new Date()); - frm.ignore_doctypes_on_cancel_all = ["Process Period Closing Voucher"]; + frm.ignore_doctypes_on_cancel_all = ["Process Period Closing Voucher", "MapReduce Job"]; }, setup: function (frm) { diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py index 9d424f9b893..851f1a0627c 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py @@ -283,6 +283,8 @@ class PeriodClosingVoucher(AccountsController): "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.mapper", "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.reducer", data, + self.doctype, + self.name, ) def on_cancel(self): @@ -292,10 +294,17 @@ class PeriodClosingVoucher(AccountsController): "Payment Ledger Entry", "Account Closing Balance", "Process Period Closing Voucher", + "MapReduce Job", ) + self.block_if_future_closing_voucher_exists() self.validate_accounts_not_frozen(for_cancellation=True) + # TODO: add branching clause based on accounts settings + from frappe.utils.background_jobs import cancel_mapreduce_job + + cancel_mapreduce_job(self.doctype, self.name) + if not frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): self.cancel_process_pcv_docs() @@ -309,6 +318,11 @@ class PeriodClosingVoucher(AccountsController): def on_trash(self): super().on_trash() + # TODO: add branching clause based on accounts settings + from frappe.utils.background_jobs import remove_mapreduce_job + + remove_mapreduce_job(self.doctype, self.name) + ppcvs = frappe.db.get_all( "Process Period Closing Voucher", {"parent_pcv": self.name, "docstatus": ["in", [1, 2]]} ) From 31205c4114b19b3090488f20844e809649ac56fb Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Tue, 1 Sep 2026 11:35:54 +0530 Subject: [PATCH 3/5] refactor: dashboard for pcv --- .../period_closing_voucher_dashboard.py | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py new file mode 100644 index 00000000000..fcbeb8350ba --- /dev/null +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher_dashboard.py @@ -0,0 +1,8 @@ +from frappe import _ + + +def get_data(): + return { + "non_standard_fieldnames": {"MapReduce Job": "document_name"}, + "transactions": [{"label": _("Job"), "items": ["MapReduce Job"]}], + } From 681bd2734fc658e27584479481951f031d2235e4 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Wed, 2 Sep 2026 13:14:16 +0530 Subject: [PATCH 4/5] refactor: post ledger entries once mapreduce is complete --- .../period_closing_voucher.py | 79 ++++++++++++++++--- 1 file changed, 66 insertions(+), 13 deletions(-) diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py index 851f1a0627c..17a62b02076 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py @@ -7,6 +7,7 @@ from datetime import timedelta import frappe from frappe import _, qb +from frappe.query_builder.custom import ConstantColumn from frappe.query_builder.functions import Max, Min, Sum from frappe.utils import ( add_days, @@ -282,6 +283,7 @@ class PeriodClosingVoucher(AccountsController): mapreduce( "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.mapper", "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.reducer", + "erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.summarize_and_post_ledger", data, self.doctype, self.name, @@ -300,12 +302,10 @@ class PeriodClosingVoucher(AccountsController): self.block_if_future_closing_voucher_exists() self.validate_accounts_not_frozen(for_cancellation=True) - # TODO: add branching clause based on accounts settings - from frappe.utils.background_jobs import cancel_mapreduce_job - - cancel_mapreduce_job(self.doctype, self.name) - if not frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): + from frappe.utils.background_jobs import cancel_mapreduce_job + + cancel_mapreduce_job(self.doctype, self.name) self.cancel_process_pcv_docs() self.db_set("gle_processing_status", "In Progress") @@ -318,10 +318,10 @@ class PeriodClosingVoucher(AccountsController): def on_trash(self): super().on_trash() - # TODO: add branching clause based on accounts settings - from frappe.utils.background_jobs import remove_mapreduce_job + if not frappe.get_single_value("Accounts Settings", "use_legacy_controller_for_pcv"): + from frappe.utils.background_jobs import remove_mapreduce_job - remove_mapreduce_job(self.doctype, self.name) + remove_mapreduce_job(self.doctype, self.name) ppcvs = frappe.db.get_all( "Process Period Closing Voucher", {"parent_pcv": self.name, "docstatus": ["in", [1, 2]]} @@ -804,7 +804,6 @@ def mapper(val): "Account", filters={"company": company, "report_type": report_type}, pluck="name" ) - # summarize gle = qb.DocType("GL Entry") query = qb.from_(gle).select(gle.account) for dim in dimensions: @@ -816,6 +815,8 @@ def mapper(val): Sum(gle.credit_in_account_currency).as_("credit_in_account_currency"), # account_currency is constant per grouped account -> Max() keeps the GROUP BY postgres-valid Max(gle.account_currency).as_("account_currency"), + ConstantColumn(balance_type).as_("balance_type"), + ConstantColumn(report_type).as_("report_type"), ).where( (gle.company.eq(company)) & (gle.is_cancelled.eq(0)) @@ -841,12 +842,10 @@ def reducer(final, partial_res): if final is None: final = [] - gl_entries = [] if partial_res: - for x in partial_res: - gl_entries.append(frappe._dict(x)) + final.extend([frappe._dict(x) for x in partial_res]) - return final + gl_entries + return final def get_dimensions(): @@ -857,3 +856,57 @@ def get_dimensions(): default_dimensions = ["cost_center", "finance_book", "project"] dimensions = default_dimensions + get_accounting_dimensions() return dimensions + + +def summarize_and_post_ledger(result, ref_dt, ref_dn): + pcv = frappe.get_doc(ref_dt, ref_dn) + + from erpnext.accounts.doctype.process_period_closing_voucher.process_period_closing_voucher import ( + build_dimension_wise_balance_dict, + get_bs_closing_entries, + get_closing_account_closing_entry, + get_gle_for_closing_account, + get_gle_for_pl_account, + get_p_l_closing_entries, + ) + + result = [frappe._dict(x) for x in result] + + # generate and post closing entries for P&L accounts + pl_entries = [x for x in result if x.report_type == "Profit and Loss"] + pl_dimension_wise_acc_balance = build_dimension_wise_balance_dict(pl_entries) + + # build gl map + pl_accounts_reverse_gle = [] + closing_account_gle = [] + + for dimensions, account_balances in pl_dimension_wise_acc_balance.items(): + for acc, balances in account_balances.items(): + balance_in_company_currency = flt(balances.debit) - flt(balances.credit) + if balance_in_company_currency: + pl_accounts_reverse_gle.append(get_gle_for_pl_account(pcv, acc, balances, dimensions)) + + closing_account_gle.append(get_gle_for_closing_account(pcv, account_balances["balances"], dimensions)) + + gl_entries = pl_accounts_reverse_gle + closing_account_gle + if gl_entries: + from erpnext.accounts.general_ledger import make_gl_entries + + make_gl_entries(gl_entries, merge_entries=False) + + # generate and post account closing balance for balance sheet accounts + bs_entries = [x for x in result if x.report_type == "Balance Sheet"] + bs_dimension_wise_acc_balance = build_dimension_wise_balance_dict(bs_entries) + pl_closing_entries = get_p_l_closing_entries(pl_accounts_reverse_gle, pcv) + bs_closing_entries = get_bs_closing_entries(bs_dimension_wise_acc_balance, pcv) + closing_entries_for_closing_account = get_closing_account_closing_entry(closing_account_gle, pcv) + closing_entries = pl_closing_entries + bs_closing_entries + closing_entries_for_closing_account + + make_closing_entries(closing_entries, pcv.name, pcv.company, pcv.period_end_date) + + # keep transaction on PPCV and PPCVD short + # prevents concurrency errors - REPEATABLE READ + if not frappe.in_test: + frappe.db.commit() # nosemgrep + + frappe.db.set_value("Period Closing Voucher", pcv.name, "gle_processing_status", "Completed") From 3f422f8e0d61b30b2705d3ae6bbea241cbee8759 Mon Sep 17 00:00:00 2001 From: ruthra kumar Date: Mon, 7 Sep 2026 16:35:06 +0530 Subject: [PATCH 5/5] refactor: use qb for estimation and include correction factor --- .../period_closing_voucher.py | 88 +++++++++++++------ 1 file changed, 63 insertions(+), 25 deletions(-) diff --git a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py index 17a62b02076..83c1db7e7d2 100644 --- a/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py +++ b/erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py @@ -3,12 +3,11 @@ import copy -from datetime import timedelta import frappe from frappe import _, qb from frappe.query_builder.custom import ConstantColumn -from frappe.query_builder.functions import Max, Min, Sum +from frappe.query_builder.functions import Count, Max, Min, Sum from frappe.utils import ( add_days, ceil, @@ -664,17 +663,41 @@ class PeriodClosingVoucher(AccountsController): def generate_tasks_for_normal_balance(self): # estimation can be wrong by a factor of 2 - estimated_count = ( - cint( - frappe.db.sql( - f"explain select count(*) from `tabGL Entry` where is_cancelled = 0 and posting_date between {self.period_start_date} and {self.period_end_date};", - as_dict=True, - )[0].rows + gl = qb.DocType("GL Entry") + raw_query = ( + qb.from_(gl) + .select(Count(gl.star)) + .where( + gl.is_cancelled.eq(0) & gl.posting_date.between(self.period_start_date, self.period_end_date) ) - * 2 + .get_sql() ) + + # estimation can be wrong by a factor of 2 + correction_factor = 2 + if frappe.db.db_type == "postgres": + analyzer = frappe.json.loads( + ( + frappe.db.sql( + f"explain (format json) {raw_query}", + ) + )[0][0] + ) + + estimated_count = analyzer[0].get("Plan").get("Plans")[0].get("Plan Rows") * correction_factor + else: + estimated_count = ( + cint( + frappe.db.sql( + f"explain {raw_query}", + as_dict=True, + )[0].rows + ) + * correction_factor + ) + job_count = ( - 1 if estimated_count / 2000000 else ceil(estimated_count / 2000000) + 1 if estimated_count / 2000000 < 1 else ceil(estimated_count / 2000000) ) # conservative chunk size days = (getdate(self.period_end_date) - getdate(self.period_start_date)).days step_size = 1 if days / job_count < 1 else ceil(days / job_count) @@ -691,18 +714,38 @@ class PeriodClosingVoucher(AccountsController): min = qb.from_(gl).select(Min(gl.posting_date)).run()[0][0] max = qb.from_(gl).select(Max(gl.posting_date)).run()[0][0] - # estimation can be wrong by a factor of 2 - estimated_count = ( - cint( - frappe.db.sql( - f"explain select count(*) from `tabGL Entry` where is_cancelled = 0 and is_opening = 0 and posting_date between {min} and {max};", - as_dict=True, - )[0].rows - ) - * 2 + raw_query = ( + qb.from_(gl) + .select(Count(gl.star)) + .where(gl.is_cancelled.eq(0) & gl.is_opening.eq("Yes") & gl.posting_date.between(min, max)) + .get_sql() ) + + # estimation can be wrong by a factor of 2 + correction_factor = 2 + if frappe.db.db_type == "postgres": + analyzer = frappe.json.loads( + ( + frappe.db.sql( + f"explain (format json) {raw_query}", + ) + )[0][0] + ) + + estimated_count = analyzer[0].get("Plan").get("Plans")[0].get("Plan Rows") * correction_factor + else: + estimated_count = ( + cint( + frappe.db.sql( + f"explain {raw_query};", + as_dict=True, + )[0].rows + ) + * correction_factor + ) + job_count = ( - 1 if estimated_count / 2000000 else ceil(estimated_count / 2000000) + 1 if estimated_count / 2000000 < 1 else ceil(estimated_count / 2000000) ) # conservative chunk size days = (getdate(self.period_end_date) - getdate(self.period_start_date)).days step_size = 1 if days / job_count < 1 else ceil(days / job_count) @@ -904,9 +947,4 @@ def summarize_and_post_ledger(result, ref_dt, ref_dn): make_closing_entries(closing_entries, pcv.name, pcv.company, pcv.period_end_date) - # keep transaction on PPCV and PPCVD short - # prevents concurrency errors - REPEATABLE READ - if not frappe.in_test: - frappe.db.commit() # nosemgrep - frappe.db.set_value("Period Closing Voucher", pcv.name, "gle_processing_status", "Completed")