test(stock): expect the live report for ageing columns in snapshot mode

Parity cases no longer ask for ageing columns, a new test checks that such a
request never opens the snapshot, and the live-test replay skips the one test that
requests them.

erpnext/stock/report/stock_balance/stock_balance.json

@@ -5,9 +5,15 @@
  "disabled": 0,
  "docstatus": 0,
  "doctype": "Report",
+ "doctype_to_sync": [
+  {
+   "doc_type": "Stock Ledger Entry"
+  }
+ ],
+ "snapshot_report": 0,
  "idx": 2,
  "is_standard": "Yes",
- "modified": "2020-04-30 13:46:14.680354",
+ "modified": "2026-09-15 12:00:00.000000",
  "modified_by": "Administrator",
  "module": "Stock",
  "name": "Stock Balance",
@@ -25,4 +31,4 @@
    "role": "Accounts Manager"
   }
  ]
-}
+}

erpnext/stock/report/stock_balance/stock_balance.py

@@ -21,6 +21,7 @@ from erpnext.stock.report.stock_ageing.stock_ageing import (
 	get_average_age,
 	normalize_fifo_queue,
 )
+from erpnext.stock.report.stock_report_snapshot import StockReportSnapshot
 from erpnext.stock.utils import add_additional_uom_columns
 
 
@@ -45,6 +46,18 @@ def execute(filters: StockBalanceFilter | None = None):
 	return StockBalanceReport(filters).run()
 
 
+def execute_snapshot_report(filters):
+	"""Ageing columns replay the ledger with live serial and batch lookups, which a frozen
+	ledger cannot match, so they keep the live report."""
+	from erpnext.stock.report.stock_balance.stock_balance_snapshot import StockBalanceSnapshotReport
+
+	if filters.get("show_stock_ageing_data"):
+		return execute(filters)
+
+	with StockReportSnapshot("Stock Balance", filters) as snapshot:
+		return StockBalanceSnapshotReport(filters, snapshot).run()
+
+
 class StockBalanceReport:
 	def __init__(self, filters: StockBalanceFilter | None) -> None:
 		self.filters = filters
@@ -216,6 +229,9 @@ class StockBalanceReport:
 			self.item_warehouse_map, self.float_precision, self.inventory_dimensions
 		)
 
+	def run_query(self, query, **kwargs):
+		return query.run(**kwargs)
+
 	def prepare_stock_reco_voucher_wise_count(self):
 		self.stock_reco_voucher_wise_count = frappe._dict()
 
@@ -266,7 +282,7 @@ class StockBalanceReport:
 			if childrens:
 				query = query.where(doctype.warehouse.isin(childrens))
 
-		data = query.run(as_dict=True)
+		data = self.run_query(query, as_dict=True)
 		if not data:
 			return
 

erpnext/stock/report/stock_balance/stock_balance_snapshot.py

@@ -0,0 +1,175 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+import frappe
+from frappe.query_builder import Case, CustomFunction
+from frappe.query_builder.functions import Abs, Cast, Function, IfNull, Min, Sum
+from pypika.analytics import CURRENT_ROW, Preceding, RowNumber
+from pypika.analytics import Sum as WindowSum
+
+from erpnext.stock.report.stock_balance.stock_balance import (
+	StockBalanceReport,
+	filter_items_with_no_transactions,
+)
+
+ArgMaxNull = CustomFunction("arg_max_null", ["value", "order"])
+ListConcat = CustomFunction("list_concat", ["left", "right"])
+
+MOVEMENT_PREFIXES = ("opening", "in", "out", "bal")
+MOVEMENT_SUFFIXES = ("qty", "val")
+MOVEMENT_FIELDS = tuple(f"{prefix}_{suffix}" for suffix in MOVEMENT_SUFFIXES for prefix in MOVEMENT_PREFIXES)
+
+
+class StockBalanceSnapshotReport(StockBalanceReport):
+	"""Stock Balance read from a DuckDB snapshot of the ledger.
+
+	Ordinary movements are summed per stock group and dimension key in DuckDB, between the
+	reconciliation rows that reset a balance; those rows and everything else go through the
+	live report's own methods. Ageing columns are served by the live report instead.
+	"""
+
+	def __init__(self, filters, snapshot):
+		super().__init__(filters)
+		self.snapshot = snapshot
+
+	def run_query(self, query, **kwargs):
+		return self.snapshot.run(query, **kwargs)
+
+	def prepare_item_warehouse_map_for_current_period(self):
+		self.opening_vouchers = self.get_opening_vouchers()
+		for row in self.snapshot.run(get_balance_query(self), as_dict=True, as_iterator=True):
+			apply_segment(self, row)
+		self.item_warehouse_map = filter_items_with_no_transactions(
+			self.item_warehouse_map, self.float_precision, self.inventory_dimensions
+		)
+
+
+def get_balance_query(report):
+	entries = frappe.qb.Table("snapshot_entries")
+	segments = frappe.qb.Table("snapshot_segments")
+	group_fields = ("item_code", "warehouse", "snapshot_dimensions")
+	segment_query = frappe.qb.from_(entries).select(
+		entries.star,
+		WindowSum(entries.snapshot_detail)
+		.over(*(entries[field] for field in group_fields))
+		.orderby(entries.snapshot_row)
+		.rows(Preceding(), CURRENT_ROW)
+		.as_("snapshot_segment"),
+	)
+	# Each detail row starts a segment and forms its own group, separate from ordinary movements.
+	return (
+		frappe.qb.with_(get_segment_query(report), "snapshot_entries")
+		.with_(segment_query, "snapshot_segments")
+		.from_(segments)
+		.select(
+			segments.item_code,
+			segments.warehouse,
+			segments.snapshot_detail,
+			Min(segments.snapshot_row).as_("snapshot_row"),
+			*get_latest_columns(report, segments),
+			*get_movement_columns(segments),
+		)
+		.groupby(
+			*(segments[field] for field in group_fields), segments.snapshot_segment, segments.snapshot_detail
+		)
+		.orderby("snapshot_row")
+	)
+
+
+def get_movement_columns(segments):
+	opening = segments.snapshot_opening
+	amounts = {
+		"qty": Cast(segments.actual_qty, "DOUBLE"),
+		"val": Cast(segments.stock_value_difference, "DOUBLE"),
+	}
+	columns = []
+	for suffix in MOVEMENT_SUFFIXES:
+		field = amounts[suffix]
+		bodies = {
+			"opening": Case().when(opening == 1, field).else_(0),
+			"in": Case().when((opening == 0) & (field >= 0), field).else_(0),
+			"out": Case().when((opening == 0) & (field < 0), -field).else_(0),
+			"bal": field,
+		}
+		columns.extend(Sum(bodies[prefix]).as_(f"{prefix}_{suffix}") for prefix in MOVEMENT_PREFIXES)
+	return columns
+
+
+def get_segment_query(report):
+	ledger = frappe.qb.DocType("Stock Ledger Entry")
+	opening = ledger.posting_date < report.from_date
+	for voucher_type, vouchers in report.opening_vouchers.items():
+		if vouchers:
+			known = report.snapshot.publish_keys(vouchers)
+			opening |= (ledger.voucher_type == voucher_type) & ledger.voucher_no.isin(known)
+
+	# flt() can classify tiny negative amounts as incoming. Preserve its exact rounding rules.
+	detail = ledger.voucher_type == "Stock Reconciliation"
+	for field in (ledger.actual_qty, ledger.stock_value_difference):
+		detail |= (field < 0) & (Abs(field) < 10**-report.float_precision)
+
+	return report.sle_query.select(
+		get_dimension_key(report, ledger).as_("snapshot_dimensions"),
+		RowNumber().orderby(ledger.posting_datetime, ledger.creation).as_("snapshot_row"),
+		Case().when(opening, 1).else_(0).as_("snapshot_opening"),
+		Case().when(detail, 1).else_(0).as_("snapshot_detail"),
+	)
+
+
+def get_dimension_key(report, ledger):
+	"""Match get_group_by_key(), including its omission of empty dimension values."""
+	key = Function("list_value")
+	for field in report.inventory_dimensions:
+		if report.filters.get(field) or report.filters.get("show_dimension_wise_stock"):
+			value = ledger[field]
+			key = ListConcat(
+				key,
+				Case()
+				.when(IfNull(value, "") != "", Function("list_value", value))
+				.else_(Function("list_value")),
+			)
+	return key
+
+
+def get_latest_columns(report, segments):
+	fields = [
+		"company",
+		"item_group",
+		"stock_uom",
+		"item_name",
+		"valuation_rate",
+		*report.inventory_dimensions,
+		"posting_date",
+		"actual_qty",
+		"stock_value_difference",
+		"voucher_type",
+		"voucher_no",
+		"batch_no",
+		"serial_no",
+		"serial_and_batch_bundle",
+		"qty_after_transaction",
+		"stock_value",
+		"voucher_detail_no",
+	]
+	return [ArgMaxNull(segments[field], segments.snapshot_row).as_(field) for field in fields]
+
+
+def apply_segment(report, row):
+	key = report.get_group_by_key(row)
+	if key not in report.item_warehouse_map:
+		report.initialize_data(key, row)
+
+	if row.snapshot_detail:
+		if row.voucher_type == "Stock Reconciliation" and not hasattr(
+			report, "stock_reco_voucher_wise_count"
+		):
+			report.prepare_stock_reco_voucher_wise_count()
+		report.prepare_item_warehouse_map(row, key)
+		return
+
+	balance = report.item_warehouse_map[key]
+	for field in MOVEMENT_FIELDS:
+		balance[field] += row[field]
+	balance.val_rate = row.valuation_rate
+	for field in report.inventory_dimensions:
+		balance[field] = row.get(field)

erpnext/stock/report/stock_balance/test_stock_balance_snapshot_report.py

@@ -0,0 +1,137 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+from copy import deepcopy
+from functools import partial
+from unittest.mock import patch
+
+import frappe
+from frappe.utils import add_days, today
+
+from erpnext.stock.report.stock_balance import stock_balance
+from erpnext.stock.report.stock_balance import test_stock_balance as live_tests
+from erpnext.stock.report.stock_report_snapshot import StockReportSnapshot
+from erpnext.stock.report.stock_snapshot_test_utils import (
+	StockSnapshotReportMixin,
+	StockSnapshotTestCase,
+	execute_on_snapshot,
+	on_snapshot,
+)
+
+
+class TestStockBalanceSnapshotReport(StockSnapshotReportMixin, StockSnapshotTestCase):
+	report = stock_balance
+	receipt_options = ({},)
+
+	def test_stock_closing_balance_matches(self):
+		self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		with patch("erpnext.stock.doctype.stock_closing_entry.stock_closing_entry.enqueue"):
+			closing = frappe.get_doc(
+				doctype="Stock Closing Entry",
+				company=self.filters.company,
+				from_date=add_days(today(), -10),
+				to_date=add_days(today(), -6),
+			).submit()
+		closing.create_stock_closing_balance_entries()
+		closing.db_set("status", "Completed")
+		self.make_movement(qty=5, basic_rate=100)
+		self.assert_snapshot_matches(stock_balance)
+
+	def test_ageing_columns_use_the_live_report(self):
+		self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		filters = frappe._dict(self.filters, show_stock_ageing_data=1)
+		with patch.object(
+			StockReportSnapshot, "get_connection", side_effect=AssertionError("snapshot opened")
+		):
+			self.assertEqual(
+				stock_balance.execute_snapshot_report(deepcopy(filters)),
+				stock_balance.execute(deepcopy(filters)),
+			)
+
+	def test_small_negative_amounts_use_existing_rounding(self):
+		self.make_movement(qty=10, basic_rate=100)
+		entry = self.make_movement(qty=1, from_warehouse="Stores - _TC", to_warehouse=None)
+		frappe.db.set_value(
+			"Stock Ledger Entry",
+			{"voucher_no": entry.name},
+			{"actual_qty": -0.00001, "stock_value_difference": -0.00001},
+		)
+		self.assert_snapshot_matches(stock_balance)
+
+	def test_movements_are_summed_as_floats(self):
+		receipt = self.make_movement(qty=1, basic_rate=100)
+		issue = self.make_movement(qty=1, from_warehouse="Stores - _TC", to_warehouse=None)
+		for entry, amount in ((receipt, 471636.5443), (issue, -459509.9528)):
+			frappe.db.set_value(
+				"Stock Ledger Entry",
+				{"voucher_no": entry.name},
+				{"actual_qty": amount, "stock_value_difference": amount},
+			)
+		self.assert_snapshot_matches(stock_balance)
+
+	def test_stock_balance_aggregates_before_python(self):
+		self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		self.make_movement(qty=5, basic_rate=150)
+		expected = stock_balance.execute(deepcopy(self.filters))
+		with patch.object(stock_balance.StockBalanceReport, "prepare_item_warehouse_map") as process_entry:
+			self.assertEqual(expected, self.run_snapshot(stock_balance, self.capture_ledger()))
+			process_entry.assert_not_called()
+
+	def test_balance_aggregates_sparse_inventory_dimensions(self):
+		for index, (project, detail) in enumerate((("A", ""), ("", "A"), ("B", "A"), (None, None), ("", ""))):
+			entry = self.make_movement(
+				qty=index + 1, basic_rate=100, posting_date=add_days(today(), index - 4)
+			)
+			frappe.db.set_value(
+				"Stock Ledger Entry",
+				{"voucher_no": entry.name},
+				{"project": project, "voucher_detail_no": detail},
+			)
+		dimensions = [
+			frappe._dict(fieldname=field, doctype="Project") for field in ("project", "voucher_detail_no")
+		]
+		with patch.object(stock_balance, "get_inventory_dimensions", return_value=dimensions):
+			for dimension_filters in ({"show_dimension_wise_stock": 1}, {"project": ["A", "B"]}, {}):
+				filters = frappe._dict(self.filters, **dimension_filters)
+				expected = stock_balance.execute(deepcopy(filters))
+				with patch.object(
+					stock_balance.StockBalanceReport,
+					"prepare_item_warehouse_map",
+					side_effect=AssertionError("Unexpected ledger replay"),
+				):
+					self.assertEqual(
+						expected, self.run_snapshot(stock_balance, self.capture_ledger(), filters)
+					)
+
+	def test_serial_bundle_details_match(self):
+		self.set_item("_Test DuckDB Serial Item", {"has_serial_no": 1, "serial_no_series": "DUCK-SN-.#####"})
+		self.make_movement(qty=3, basic_rate=100)
+		self.assert_snapshot_matches(stock_balance)
+
+	def test_batch_opening_and_bundle_details_match(self):
+		self.make_batch_history()
+		self.assert_snapshot_matches(stock_balance)
+
+	def test_inventory_dimension_opening_and_grouping_match(self):
+		opening = self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		current = self.make_movement(qty=5, basic_rate=100)
+		for entry in (opening, current):
+			frappe.db.set_value("Stock Ledger Entry", {"voucher_no": entry.name}, "project", "DuckDB Project")
+		dimensions = [frappe._dict(fieldname="project", doctype="Project")]
+		with patch.object(stock_balance, "get_inventory_dimensions", return_value=dimensions):
+			self.assert_snapshot_matches(
+				stock_balance, project=["DuckDB Project"], show_dimension_wise_stock=1
+			)
+
+	def test_item_group_and_brand_filters(self):
+		self.make_movement(qty=10, basic_rate=100)
+		parent = frappe.get_value("Warehouse", "Stores - _TC", "parent_warehouse")
+		self.assert_snapshot_matches(stock_balance, warehouse=[parent], item_group="Products")
+		self.assert_snapshot_matches(stock_balance, brand="No matching brand")
+
+
+TestStockBalanceOnSnapshot = on_snapshot(
+	live_tests.TestStockBalance,
+	skip=("test_show_stock_ageing_data_adds_ageing_columns",),
+	execute=partial(execute_on_snapshot, stock_balance),
+)

erpnext/stock/report/stock_report_snapshot.py

@@ -0,0 +1,331 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+from contextlib import ExitStack, closing
+from copy import copy
+from datetime import timedelta
+from itertools import batched
+from operator import itemgetter
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from typing import NamedTuple
+
+import frappe
+from frappe import _
+from frappe.query_builder import CustomFunction
+from frappe.query_builder.functions import Cast
+from frappe.query_builder.terms import NamedParameterWrapper
+from pypika.terms import Star
+
+BATCH_SIZE = 1000
+MAX_LIVE_TABLE_BYTES = 64 * 1024 * 1024
+
+
+class StockReportSnapshot:
+	"""Run stock queries against synced ledger entries and live supporting records.
+
+	Only Stock Ledger Entry comes from the snapshot. Every doctype in LIVE_TABLES is read from
+	the live database the first time a query needs it, so a report combines a frozen ledger with
+	current master data. Those rows cover the ledger keys under the report's company, item and
+	to_date filters, so a query that joins one of them must stay inside the same ledger scope.
+	"""
+
+	def __init__(self, report_name, filters=None):
+		self.conn = self.get_connection(report_name)
+		self.tables = {}
+		self.filters = filters or {}
+		self.temp_directory = None
+
+	def __enter__(self):
+		return self
+
+	def __exit__(self, *exc):
+		try:
+			self.conn.close()
+		finally:
+			self.tables.clear()
+			if self.temp_directory:
+				self.temp_directory.cleanup()
+
+	@staticmethod
+	def get_connection(report_name):
+		"""Open the latest submitted sync. An older complete sync is not served because the desk
+		labels snapshot results with the latest submitted sync's timestamp."""
+		sync = frappe.db.get_value(
+			"DuckDB Sync",
+			{"doc_type": "Stock Ledger Entry", "docstatus": 1},
+			"name",
+			order_by="creation desc",
+		)
+		if not sync or frappe.db.exists("DuckDB Sync Item", {"parent": sync, "synced": 0}):
+			frappe.throw(
+				_("{0} requires a completed Stock Ledger Entry sync to DuckDB").format(_(report_name))
+			)
+		return frappe.get_doc("DuckDB Sync", sync).get_duckdb_conn()
+
+	def run(self, query, as_dict=False, as_iterator=False, pluck=False):
+		cursor = self.execute_query(query, convert=True)
+		rows = self.iter_rows(cursor, as_dict, pluck)
+		return rows if as_iterator else list(rows)
+
+	def execute_query(self, query, convert=False):
+		sql, parameters = self.compile(query)
+		cursor = self.conn.cursor()
+		try:
+			for name in self.get_referenced_tables(query):
+				cursor.register(name, self.get_table(name))
+			if convert:
+				sql, parameters = self.convert_result_types(cursor, query, sql, parameters)
+			cursor.execute(sql, parameters)
+		except Exception:
+			cursor.close()
+			raise
+
+		return cursor
+
+	def convert_result_types(self, cursor, query, sql, parameters):
+		"""Return decimals as doubles and times as intervals, so Python receives the floats and
+		timedeltas the live database returns without converting every value itself. DESCRIBE
+		binds the statement without running it."""
+		description = cursor.execute(f"DESCRIBE {sql}", parameters).fetchall()
+		columns = [row[0] for row in description]
+		conversions = [
+			(index, base_type(row[1]))
+			for index, row in enumerate(description)
+			if base_type(row[1]) in CONVERTED_TERMS
+		]
+		if not conversions:
+			return sql, parameters
+		casted = cast_selected_columns(query, conversions)
+		if casted is not None:
+			return self.compile(casted)
+		if has_repeated_names(columns):
+			return sql, parameters
+		replaced = []
+		for index, dtype in conversions:
+			name = '"' + columns[index].replace('"', '""') + '"'
+			replaced.append(f"{CONVERTED_TYPES[dtype](name)} AS {name}")
+		return f"SELECT * REPLACE ({', '.join(replaced)}) FROM ({sql}) AS converted", parameters
+
+	@staticmethod
+	def compile(query):
+		parameters = DuckDBParameters()
+		sql = query.get_sql(quote_char='"', alias_quote_char='"', param_wrapper=parameters)
+		return sql, parameters.get_parameters()
+
+	def get_referenced_tables(self, query) -> list[str]:
+		# DuckDB's dependency parser does not accept prepared parameters. Only inspection uses
+		# literal values rendered by the query builder; run() executes with bound parameters.
+		sql = query.get_sql(quote_char='"', alias_quote_char='"')
+		referenced = self.conn.get_table_names(sql)
+		names = dict.fromkeys([*self.tables, *(f"tab{doctype}" for doctype in LIVE_TABLES)])
+		return [name for name in names if name in referenced]
+
+	def register(self, name, rows, fields):
+		"""Expose a lookup table built in Python to the queries that follow."""
+		import pyarrow as pa
+
+		self.tables[name] = pa.Table.from_pylist(rows, schema=arrow_schema(fields))
+
+	def publish_keys(self, values):
+		"""A lookup table of the values, as the subquery to match a column against."""
+		name = f"snapshot_keys_{len(self.tables)}"
+		self.register(name, [{"key": value} for value in dict.fromkeys(values)], {"key": "string"})
+		keys = frappe.qb.Table(name)
+		return frappe.qb.from_(keys).select(keys.key)
+
+	def get_table(self, name):
+		if name not in self.tables:
+			self.tables[name] = self.build_live_table(name.removeprefix("tab"))
+		return self.tables[name]
+
+	def build_live_table(self, doctype):
+		"""Keep small lookups in memory and spill large ones to a reusable Arrow file."""
+		import pyarrow as pa
+
+		schema = arrow_schema(LIVE_TABLES[doctype].fields)
+		batches, size = [], 0
+		writer = None
+		with ExitStack() as stack:
+			reader = stack.enter_context(closing(self.iter_live_batches(doctype, schema)))
+			for batch in reader:
+				if writer is None:
+					batches.append(batch)
+					size += batch.nbytes
+					if size <= MAX_LIVE_TABLE_BYTES:
+						continue
+					if self.temp_directory is None:
+						self.temp_directory = TemporaryDirectory(prefix="stock-report-")
+					path = Path(self.temp_directory.name) / f"{doctype}.arrow"
+					writer = stack.enter_context(pa.ipc.new_file(str(path), schema))
+					for buffered in batches:
+						writer.write_batch(buffered)
+					batches.clear()
+				else:
+					writer.write_batch(batch)
+
+		if writer is not None:
+			import pyarrow.dataset as ds
+
+			return ds.dataset(path, format="ipc")
+		return pa.Table.from_batches(batches, schema=schema)
+
+	def iter_live_batches(self, doctype, schema):
+		import pyarrow as pa
+
+		table = LIVE_TABLES[doctype]
+		with closing(self.get_ledger_values(table.ledger_field)) as names:
+			for keys in batched(names, BATCH_SIZE):
+				filters = {table.link_field: ("in", list(keys))}
+				query = frappe.get_all(doctype, filters=filters, fields=list(table.fields), run=False)
+				with frappe.db.unbuffered_cursor():
+					rows = query.run(as_dict=True, as_iterator=True)
+					for batch in batched(rows, BATCH_SIZE):
+						yield pa.RecordBatch.from_pylist(batch, schema=schema)
+
+	def get_ledger_values(self, field):
+		"""Ledger keys a supporting table has to cover, under the report's own ledger scope."""
+		ledger = frappe.qb.DocType("Stock Ledger Entry")
+		query = frappe.qb.from_(ledger).select(ledger[field]).distinct().where(ledger[field].notnull())
+		for key in ("company", "item_code"):
+			if value := self.filters.get(key):
+				values = value if isinstance(value, list | tuple) else [value]
+				query = query.where(ledger[key].isin(values))
+		if to_date := self.filters.get("to_date"):
+			query = query.where(ledger.posting_date <= to_date)
+		with closing(self.run(query, pluck=True, as_iterator=True)) as rows:
+			yield from (value for value in rows if value)
+
+	@staticmethod
+	def iter_rows(cursor, as_dict, pluck):
+		columns = [column[0] for column in cursor.description]
+		# Resolve conversions once per column, rather than inspecting every value in Python.
+		converters = [
+			(index, converter)
+			for index, column in enumerate(cursor.description)
+			if (converter := VALUE_CONVERTERS.get(column[1].id))
+		]
+		if pluck:
+			shape = itemgetter(0)
+		elif as_dict:
+
+			def shape(row):
+				return frappe._dict(zip(columns, row, strict=True))
+		else:
+			shape = tuple
+
+		try:
+			while rows := cursor.fetchmany(1000):
+				for row in rows:
+					if converters:
+						row = list(row)
+						for index, converter in converters:
+							if row[index] is not None:
+								row[index] = converter(row[index])
+					yield shape(row)
+		finally:
+			cursor.close()
+
+
+def base_type(dtype):
+	return str(dtype).split("(")[0]
+
+
+def cast_selected_columns(query, conversions):
+	"""The query with its converted columns cast in the select list, which DuckDB streams, or None
+	when the select list does not map onto the result columns one to one."""
+	selects = getattr(query, "_selects", None)
+	if not selects or any(isinstance(term, Star) for term in selects):
+		return None
+	selects = list(selects)
+	for index, dtype in conversions:
+		if index >= len(selects):
+			return None
+		term = selects[index]
+		name = term.alias or getattr(term, "name", None)
+		if not name:
+			return None
+		selects[index] = CONVERTED_TERMS[dtype](term).as_(name)
+	casted = copy(query)
+	casted._selects = selects
+	return casted
+
+
+def has_repeated_names(columns):
+	"""A repeated column comes back suffixed with _1, _2 ... once a subquery binds it, which
+	would change the keys a report reads; such results keep the live column names instead."""
+	names = set(columns)
+	for column in columns:
+		base, _, suffix = column.rpartition("_")
+		if suffix.isdigit() and base in names:
+			return True
+	return False
+
+
+def arrow_schema(fields):
+	import pyarrow as pa
+
+	return pa.schema([(field, getattr(pa, dtype)()) for field, dtype in fields.items()])
+
+
+def time_to_timedelta(value):
+	return timedelta(
+		hours=value.hour, minutes=value.minute, seconds=value.second, microseconds=value.microsecond
+	)
+
+
+VALUE_CONVERTERS = {"decimal": float, "time": time_to_timedelta, "time_tz": time_to_timedelta}
+EpochMicroseconds = CustomFunction("epoch_us", ["value"])
+ToMicroseconds = CustomFunction("to_microseconds", ["value"])
+CONVERTED_TERMS = {
+	"DECIMAL": lambda term: Cast(term, "DOUBLE"),
+	"TIME": lambda term: ToMicroseconds(EpochMicroseconds(term)),
+}
+CONVERTED_TYPES = {
+	"DECIMAL": lambda name: f"CAST({name} AS DOUBLE)",
+	"TIME": lambda name: f"to_microseconds(epoch_us({name}))",
+}
+
+
+class DuckDBParameters(NamedParameterWrapper):
+	def get_sql(self, param_value, **kwargs):
+		key = f"param{len(self.parameters) + 1}"
+		self.parameters[key] = param_value
+		return f"${key}"
+
+
+class LiveTable(NamedTuple):
+	"""A supporting doctype read from the live database for the queries that need it.
+
+	`ledger_field` is the Stock Ledger Entry column holding its keys and `link_field` the column
+	those keys match.
+	"""
+
+	ledger_field: str
+	link_field: str
+	fields: dict[str, str]
+
+
+LIVE_TABLES = {
+	"Item": LiveTable(
+		"item_code",
+		"name",
+		{
+			"name": "string",
+			"item_code": "string",
+			"item_name": "string",
+			"description": "string",
+			"stock_uom": "string",
+			"brand": "string",
+			"item_group": "string",
+			"has_serial_no": "int64",
+			"has_batch_no": "int64",
+			"valuation_method": "string",
+		},
+	),
+	"Warehouse": LiveTable(
+		"warehouse",
+		"name",
+		{"name": "string", "lft": "int64", "rgt": "int64", "warehouse_type": "string"},
+	),
+}

erpnext/stock/report/stock_snapshot_test_utils.py

@@ -0,0 +1,213 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+from contextlib import contextmanager
+from copy import deepcopy
+from datetime import time, timedelta
+from unittest.mock import patch
+
+import duckdb
+import frappe
+import pyarrow as pa
+from frappe.database.duckdb.schema import DuckDBTable
+from frappe.utils import add_days, today
+
+from erpnext.stock.doctype.item.test_item import make_item
+from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
+from erpnext.stock.doctype.stock_reconciliation.test_stock_reconciliation import create_stock_reconciliation
+from erpnext.stock.report.stock_report_snapshot import StockReportSnapshot
+from erpnext.tests.utils import ERPNextTestSuite
+
+
+class StockSnapshotTestCase(ERPNextTestSuite):
+	def setUp(self):
+		self.item = make_item("_Test DuckDB Stock Item").name
+		self.filters = frappe._dict(
+			company="_Test Company",
+			item_code=[self.item],
+			warehouse=["Stores - _TC"],
+			from_date=add_days(today(), -5),
+			to_date=today(),
+			range="30, 60, 90",
+		)
+
+	def assert_snapshot_matches(self, report, **filters):
+		report_filters = deepcopy(self.filters)
+		report_filters.update(filters)
+		expected = report.execute(deepcopy(report_filters))
+		actual = self.run_snapshot(report, self.capture_ledger(report_filters.item_code), report_filters)
+		self.assertEqual(expected, actual)
+
+	def run_snapshot(self, report, table, filters=None):
+		with snapshot_of(table):
+			return report.execute_snapshot_report(deepcopy(filters or self.filters))
+
+	def capture_ledger(self, items=None):
+		rows = frappe.get_all(
+			"Stock Ledger Entry",
+			filters={"item_code": ("in", items or [self.item])},
+			fields=frappe.get_meta("Stock Ledger Entry").get_valid_columns(),
+		)
+		for row in rows:
+			if isinstance(row.posting_time, timedelta):
+				seconds = row.posting_time.seconds
+				row.posting_time = time(seconds // 3600, seconds % 3600 // 60, seconds % 60)
+		return pa.Table.from_pylist(rows, schema=DuckDBTable("Stock Ledger Entry").get_arrow_schema())
+
+	@staticmethod
+	def connect(table):
+		conn = duckdb.connect(":memory:")
+		DuckDBTable("Stock Ledger Entry").sync(conn)
+		conn.register("snapshot_data", table)
+		source = frappe.qb.Table("snapshot_data")
+		query = (
+			frappe.qb.into(frappe.qb.DocType("Stock Ledger Entry"))
+			.columns(*table.column_names)
+			.from_(source)
+			.select(*(source[field] for field in table.column_names))
+		)
+		conn.execute(*StockReportSnapshot.compile(query))
+		conn.unregister("snapshot_data")
+		return conn
+
+	def make_movement(self, **kwargs):
+		args = dict(
+			item_code=self.item,
+			to_warehouse="Stores - _TC",
+			posting_date=today(),
+			posting_time="12:00:00",
+		)
+		args.update(kwargs)
+		return make_stock_entry(**args)
+
+	def set_item(self, name, properties):
+		self.item = make_item(name, properties).name
+		self.filters.item_code = [self.item]
+
+	@property
+	def reports(self):
+		return (self.report,)
+
+	def make_batch_history(self):
+		self.set_item(
+			"_Test DuckDB Batch Item",
+			{"has_batch_no": 1, "create_new_batch": 1, "batch_number_series": "DUCK-.#####"},
+		)
+		receipt = self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		batch = frappe.get_value(
+			"Serial and Batch Entry", {"parent": receipt.items[0].serial_and_batch_bundle}, "batch_no"
+		)
+		self.make_movement(qty=3, batch_no=batch, from_warehouse="Stores - _TC", to_warehouse=None)
+		return batch
+
+
+@contextmanager
+def snapshot_of(table):
+	"""Serve the ledger rows as the report's snapshot and refuse any live ledger query meanwhile."""
+	conn = StockSnapshotTestCase.connect(table)
+	original_sql = frappe.db.sql
+
+	def guarded_sql(query, *args, **kwargs):
+		if "tabStock Ledger Entry" in str(query):
+			raise AssertionError(f"live ledger query during a snapshot run: {query}")
+		return original_sql(query, *args, **kwargs)
+
+	with (
+		patch.object(StockReportSnapshot, "get_connection", return_value=conn),
+		patch.object(frappe.db, "sql", side_effect=guarded_sql),
+	):
+		yield conn
+
+
+def ledger_scope(filters):
+	if items := filters.get("item_code"):
+		return {"item_code": ("in", items if isinstance(items, list | tuple) else [items])}
+	if company := filters.get("company"):
+		return {"company": company}
+	return {}
+
+
+def execute_on_snapshot(report, filters):
+	"""Run a report the way the live tests call execute, but from a snapshot of the ledger."""
+	rows = frappe.get_all(
+		"Stock Ledger Entry",
+		filters=ledger_scope(filters),
+		fields=frappe.get_meta("Stock Ledger Entry").get_valid_columns(),
+	)
+	for row in rows:
+		if isinstance(row.posting_time, timedelta):
+			seconds = row.posting_time.seconds
+			row.posting_time = time(seconds // 3600, seconds % 3600 // 60, seconds % 60)
+	table = pa.Table.from_pylist(rows, schema=DuckDBTable("Stock Ledger Entry").get_arrow_schema())
+	with snapshot_of(table):
+		return report.execute_snapshot_report(deepcopy(filters))
+
+
+def on_snapshot(live_tests, skip=(), **replacements):
+	"""A copy of a live report test class that runs the report from a DuckDB snapshot instead,
+	without the tests named in `skip`."""
+
+	def setUp(self):
+		for name, replacement in replacements.items():
+			patcher = patch(f"{live_tests.__module__}.{name}", replacement)
+			patcher.start()
+			self.addCleanup(patcher.stop)
+		live_tests.setUp(self)
+
+	members = {"setUp": setUp, **dict.fromkeys(skip)}
+	return type(f"{live_tests.__name__}OnSnapshot", (live_tests,), members)
+
+
+class StockSnapshotReportMixin:
+	def test_spilled_supporting_tables_match(self):
+		self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		self.make_movement(qty=3, from_warehouse="Stores - _TC", to_warehouse=None)
+		with patch("erpnext.stock.report.stock_report_snapshot.MAX_LIVE_TABLE_BYTES", 1):
+			self.assert_snapshot_matches(self.report)
+
+	def test_snapshot_does_not_read_live_ledger_changes(self):
+		self.make_movement(qty=10, basic_rate=100)
+		snapshot = self.capture_ledger()
+		expected = {report: report.execute(deepcopy(self.filters)) for report in self.reports}
+		self.make_movement(qty=5, basic_rate=100)
+		for report in self.reports:
+			with self.subTest(report=report.__name__):
+				self.assertEqual(expected[report], self.run_snapshot(report, snapshot))
+				self.assertNotEqual(expected[report][1], report.execute(deepcopy(self.filters))[1])
+
+	def test_stock_reconciliation_matches(self):
+		self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		create_stock_reconciliation(
+			item_code=self.item,
+			warehouse="Stores - _TC",
+			qty=7,
+			valuation_rate=120,
+			posting_date=add_days(today(), -2),
+			posting_time="12:00:00",
+		)
+		self.make_movement(qty=2, basic_rate=130, posting_date=add_days(today(), -1))
+		for report in self.reports:
+			with self.subTest(report=report.__name__):
+				self.assert_snapshot_matches(report)
+
+	def test_empty_snapshot_preserves_columns(self):
+		for report in self.reports:
+			with self.subTest(report=report.__name__):
+				self.assert_snapshot_matches(report)
+
+	def test_reports_match_with_opening_receipts_and_issues(self):
+		self.make_movement(qty=10, basic_rate=100, posting_date=add_days(today(), -10))
+		self.make_movement(qty=5, basic_rate=150)
+		self.make_movement(qty=3, from_warehouse="Stores - _TC", to_warehouse=None)
+		for options in self.receipt_options:
+			self.assert_snapshot_matches(self.report, **options)
+
+	def test_filters_and_cancelled_entries_match(self):
+		self.make_movement(qty=10, basic_rate=100)
+		cancelled = self.make_movement(qty=2, basic_rate=100)
+		cancelled.cancel()
+		for report in self.reports:
+			with self.subTest(report=report.__name__):
+				self.assert_snapshot_matches(report)
+		parent = frappe.get_value("Warehouse", "Stores - _TC", "parent_warehouse")
+		self.assert_snapshot_matches(self.report, warehouse=[parent])

erpnext/stock/report/test_stock_report_snapshot.py

@@ -0,0 +1,196 @@
+# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
+# License: GNU General Public License v3. See license.txt
+
+from datetime import time, timedelta
+from decimal import Decimal
+from pathlib import Path
+from random import Random
+from unittest.mock import patch
+
+import frappe
+import pyarrow as pa
+import pyarrow.dataset as ds
+from frappe.core.doctype.duckdb_sync.duckdb_sync import DuckDBSync
+from frappe.query_builder.builder import MariaDB, Postgres
+from frappe.query_builder.functions import Cast, Count
+
+from erpnext.stock.doctype.item.test_item import make_item
+from erpnext.stock.report.stock_report_snapshot import StockReportSnapshot
+from erpnext.stock.report.stock_snapshot_test_utils import StockSnapshotTestCase
+
+
+class TestStockReportSnapshot(StockSnapshotTestCase):
+	def test_supporting_tables_are_read_in_batches(self):
+		items = [self.item, make_item("_Test DuckDB Batched Item").name]
+		self.filters.item_code = items
+		for item in items:
+			self.make_movement(item_code=item, qty=1, basic_rate=100)
+		conn = self.connect(self.capture_ledger(items))
+		with (
+			patch.object(StockReportSnapshot, "get_connection", return_value=conn),
+			patch("erpnext.stock.report.stock_report_snapshot.BATCH_SIZE", 1),
+			patch.object(frappe, "get_all", wraps=frappe.get_all) as get_all,
+		):
+			with StockReportSnapshot("Stock Balance", self.filters) as snapshot:
+				self.assertEqual(snapshot.get_table("tabItem").num_rows, 2)
+
+		self.assertEqual(get_all.call_count, 2)
+		for call in get_all.call_args_list:
+			self.assertEqual(len(call.kwargs["filters"]["name"][1]), 1)
+
+	def test_snapshot_result_types_and_nulls(self):
+		self.make_movement(qty=1, basic_rate=100.25)
+		ledger = frappe.qb.DocType("Stock Ledger Entry")
+		query = (
+			frappe.qb.from_(ledger)
+			.select(
+				ledger.incoming_rate,
+				ledger.posting_time,
+				ledger.posting_date,
+				Cast(None, "DECIMAL(21,9)").as_("empty_qty"),
+				Cast(None, "TIME").as_("empty_time"),
+				ledger.item_code,
+			)
+			.where(ledger.item_code == self.item)
+		)
+		expected = [tuple(row) for row in query.run()]
+		expected_dicts = query.run(as_dict=True)
+		conn = self.connect(self.capture_ledger())
+		with patch.object(StockReportSnapshot, "get_connection", return_value=conn):
+			with StockReportSnapshot("Stock Ledger", self.filters) as snapshot:
+				self.assertEqual(snapshot.run(query), expected)
+				self.assertEqual(snapshot.run(query, as_dict=True), expected_dicts)
+				self.assertEqual(list(snapshot.run(query, as_iterator=True)), expected)
+				self.assertEqual(snapshot.run(query, pluck=True), [100.25])
+				with patch("erpnext.stock.report.stock_report_snapshot.VALUE_CONVERTERS", {}):
+					self.assertEqual(snapshot.run(query), expected)
+				repeated = frappe.qb.from_(ledger).select(
+					ledger.item_code, ledger.item_code, ledger.incoming_rate
+				)
+				repeated = repeated.where(ledger.item_code == self.item)
+				self.assertEqual(snapshot.run(repeated, as_dict=True), repeated.run(as_dict=True))
+
+	def test_result_conversion_preserves_python_values(self):
+		random = Random(42)
+		values = [None, Decimal("999999999999.999999999"), Decimal("-0.000000001"), Decimal("0.1")]
+		values.extend(Decimal(random.randrange(-(10**21), 10**21)).scaleb(-9) for _ in range(1000))
+		times = [None, time(), time(23, 59, 59, 999999), time(12, 30, 45, 123456)] * 251
+		source = frappe.qb.Table("native_values")
+		query = frappe.qb.from_(source).select(source.actual_qty, source.posting_time)
+		with (
+			patch.object(
+				StockReportSnapshot, "get_connection", return_value=self.connect(self.capture_ledger())
+			),
+			patch("erpnext.stock.report.stock_report_snapshot.VALUE_CONVERTERS", {}),
+		):
+			with StockReportSnapshot("Stock Ledger", self.filters) as snapshot:
+				snapshot.tables["native_values"] = pa.table(
+					{
+						"actual_qty": pa.array(values, type=pa.decimal128(21, 9)),
+						"posting_time": pa.array(times, type=pa.time64("us")),
+					}
+				)
+				actual = snapshot.run(query)
+		expected = [
+			(
+				float(value) if value is not None else None,
+				timedelta(
+					hours=clock.hour,
+					minutes=clock.minute,
+					seconds=clock.second,
+					microseconds=clock.microsecond,
+				)
+				if clock is not None
+				else None,
+			)
+			for value, clock in zip(values, times, strict=True)
+		]
+		self.assertEqual(actual, expected)
+
+	def test_large_supporting_table_spills_and_can_be_scanned_repeatedly(self):
+		items = [self.item, *(make_item(f"_Test DuckDB Spill Item {index}").name for index in range(6))]
+		self.filters.item_code = items
+		for item in items:
+			self.make_movement(item_code=item, qty=1, basic_rate=100)
+		conn = self.connect(self.capture_ledger(items))
+		item = frappe.qb.DocType("Item")
+		query = frappe.qb.from_(item).select(item.name, item.brand).orderby(item.name)
+		with (
+			patch.object(StockReportSnapshot, "get_connection", return_value=conn),
+			patch("erpnext.stock.report.stock_report_snapshot.BATCH_SIZE", 2),
+			patch("erpnext.stock.report.stock_report_snapshot.MAX_LIVE_TABLE_BYTES", 1),
+			patch.object(frappe.db, "unbuffered_cursor", wraps=frappe.db.unbuffered_cursor) as unbuffered,
+		):
+			with StockReportSnapshot("Stock Balance", self.filters) as snapshot:
+				table = snapshot.get_table("tabItem")
+				self.assertIsInstance(table, ds.FileSystemDataset)
+				path = Path(table.files[0])
+				self.assertTrue(path.exists())
+				self.assertEqual([batch.num_rows for batch in table.to_batches()], [2, 2, 2, 1])
+				expected = snapshot.run(query)
+				self.assertEqual([name for name, _brand in expected], sorted(items))
+				self.assertEqual(snapshot.run(query), expected)
+				other = item.as_("other")
+				join = (
+					frappe.qb.from_(item)
+					.inner_join(other)
+					.on(item.name != other.name)
+					.select(Count(item.name))
+				)
+				self.assertEqual(snapshot.run(join), [(42,)])
+				unbuffered.assert_called()
+		self.assertFalse(path.exists())
+
+	def test_failed_spill_cleans_files_and_restores_live_cursor(self):
+		self.make_movement(qty=1, basic_rate=100)
+		conn = self.connect(self.capture_ledger())
+		original_cursor = frappe.db._cursor
+		with (
+			patch.object(StockReportSnapshot, "get_connection", return_value=conn),
+			patch("erpnext.stock.report.stock_report_snapshot.MAX_LIVE_TABLE_BYTES", 1),
+			patch("pyarrow.ipc.new_file", side_effect=OSError("disk full")),
+		):
+			with self.assertRaisesRegex(OSError, "disk full"):
+				with StockReportSnapshot("Stock Ledger", self.filters) as snapshot:
+					snapshot.get_table("tabItem")
+		self.assertFalse(Path(snapshot.temp_directory.name).exists())
+		self.assertIs(frappe.db._cursor, original_cursor)
+		self.assertEqual(frappe.db.get_value("Item", self.item), self.item)
+
+	def test_requires_latest_submitted_sync_to_be_complete(self):
+		complete = frappe.get_doc(doctype="DuckDB Sync", doc_type="Stock Ledger Entry").insert()
+		complete.db_set("docstatus", 1)
+		complete.db_tables[0].db_set("synced", 1)
+		pending = frappe.get_doc(doctype="DuckDB Sync", doc_type="Stock Ledger Entry").insert()
+		pending.db_set("docstatus", 1)
+		frappe.get_doc(doctype="DuckDB Sync", doc_type="Stock Ledger Entry").insert()
+		with patch.object(DuckDBSync, "get_duckdb_conn", autospec=True, side_effect=lambda doc: doc.name):
+			self.assertRaises(frappe.ValidationError, StockReportSnapshot.get_connection, "Stock Ledger")
+			pending.db_tables[0].db_set("synced", 1)
+			self.assertEqual(StockReportSnapshot.get_connection("Stock Ledger"), pending.name)
+
+	def test_query_parameters_and_dialects(self):
+		self.make_movement(qty=1, basic_rate=100)
+		table = self.capture_ledger()
+		with patch.object(StockReportSnapshot, "get_connection", side_effect=lambda _: self.connect(table)):
+			with StockReportSnapshot("Stock Ledger") as snapshot:
+				for builder in (MariaDB, Postgres):
+					with self.subTest(builder=builder.__name__):
+						sle = builder.DocType("Stock Ledger Entry")
+						query = builder.from_(sle).select(sle.item_code).where(sle.item_code == self.item)
+						self.assertEqual(snapshot.run(query, pluck=True), [self.item])
+						query = builder.from_(sle).select(sle.name).where(sle.item_code == "x' OR 1=1 -- `")
+						self.assertEqual(snapshot.run(query), [])
+
+	def test_cte_named_after_live_table_does_not_load_that_doctype(self):
+		self.make_movement(qty=1, basic_rate=100)
+		ledger = frappe.qb.DocType("Stock Ledger Entry")
+		item = frappe.qb.DocType("Item")
+		entries = frappe.qb.from_(ledger).select(ledger.item_code.as_("name"))
+		query = frappe.qb.with_(entries, "tabItem").from_(item).select(item.name)
+		conn = self.connect(self.capture_ledger())
+		with patch.object(StockReportSnapshot, "get_connection", return_value=conn):
+			with StockReportSnapshot("Stock Ledger", self.filters) as snapshot:
+				with patch.object(snapshot, "build_live_table") as build_live_table:
+					self.assertEqual(snapshot.run(query, pluck=True), [self.item])
+					build_live_table.assert_not_called()