Merge pull request #58732 from ruthra-kumar/rewriting_process_pcv_on_map_reduce

refactor: rewriting process pcv on mapreduce

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) {

erpnext/accounts/doctype/period_closing_voucher/period_closing_voucher.py

@@ -5,9 +5,20 @@
 import copy
 
 import frappe
-from frappe import _
+from frappe import _, qb
-from frappe.query_builder.functions import Max, Sum
+from frappe.query_builder.custom import ConstantColumn
-from frappe.utils import add_days, flt, fmt_money, formatdate, get_link_to_form, getdate
+from frappe.query_builder.functions import Count, 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,17 @@ 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})
+			from frappe.utils.background_jobs import mapreduce
-			ppcv.save().submit()
+
+			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",
+				"erpnext.accounts.doctype.period_closing_voucher.period_closing_voucher.summarize_and_post_ledger",
+				data,
+				self.doctype,
+				self.name,
+			)
 
 	def on_cancel(self):
 		self.ignore_linked_doctypes = (
@@ -275,11 +295,16 @@ 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)
 
 		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")
@@ -292,6 +317,11 @@ class PeriodClosingVoucher(AccountsController):
 
 	def on_trash(self):
 		super().on_trash()
+		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)
+
 		ppcvs = frappe.db.get_all(
 			"Process Period Closing Voucher", {"parent_pcv": self.name, "docstatus": ["in", [1, 2]]}
 		)
@@ -594,6 +624,135 @@ 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
+		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)
+			)
+			.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 < 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)
+		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]
+
+			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 < 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)
+			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 +832,119 @@ 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"
+	)
+
+	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"),
+		ConstantColumn(balance_type).as_("balance_type"),
+		ConstantColumn(report_type).as_("report_type"),
+	).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 = []
+
+	if partial_res:
+		final.extend([frappe._dict(x) for x in partial_res])
+
+	return final
+
+
+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
+
+
+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)
+
+	frappe.db.set_value("Period Closing Voucher", pcv.name, "gle_processing_status", "Completed")

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"]}],
+	}