diff --git a/erpnext/controllers/sales_and_purchase_return.py b/erpnext/controllers/sales_and_purchase_return.py
index 81e71e3..81080f0 100644
--- a/erpnext/controllers/sales_and_purchase_return.py
+++ b/erpnext/controllers/sales_and_purchase_return.py
@@ -8,6 +8,8 @@
 from frappe.utils import flt, format_datetime, get_datetime
 
 import erpnext
+from erpnext.stock.serial_batch_bundle import get_batches_from_bundle
+from erpnext.stock.serial_batch_bundle import get_serial_nos as get_serial_nos_from_bundle
 from erpnext.stock.utils import get_incoming_rate
 
 
@@ -69,8 +71,6 @@
 
 
 def validate_returned_items(doc):
-	from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
-
 	valid_items = frappe._dict()
 
 	select_fields = "item_code, qty, stock_qty, rate, parenttype, conversion_factor"
@@ -123,26 +123,6 @@
 						)
 					)
 
-				elif ref.batch_no and d.batch_no not in ref.batch_no:
-					frappe.throw(
-						_("Row # {0}: Batch No must be same as {1} {2}").format(
-							d.idx, doc.doctype, doc.return_against
-						)
-					)
-
-				elif ref.serial_no:
-					if d.qty and not d.serial_no:
-						frappe.throw(_("Row # {0}: Serial No is mandatory").format(d.idx))
-					else:
-						serial_nos = get_serial_nos(d.serial_no)
-						for s in serial_nos:
-							if s not in ref.serial_no:
-								frappe.throw(
-									_("Row # {0}: Serial No {1} does not match with {2} {3}").format(
-										d.idx, s, doc.doctype, doc.return_against
-									)
-								)
-
 				if (
 					warehouse_mandatory
 					and not d.get("warehouse")
@@ -397,71 +377,92 @@
 		else:
 			doc.run_method("calculate_taxes_and_totals")
 
-	def update_item(source_doc, target_doc, source_parent):
+	def update_serial_batch_no(source_doc, target_doc, source_parent, item_details, qty_field):
+		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
 		from erpnext.stock.serial_batch_bundle import SerialBatchCreation
 
-		target_doc.qty = -1 * source_doc.qty
-		item_details = frappe.get_cached_value(
-			"Item", source_doc.item_code, ["has_batch_no", "has_serial_no"], as_dict=1
-		)
-
 		returned_serial_nos = []
-		if source_doc.get("serial_and_batch_bundle"):
-			if item_details.has_serial_no:
-				returned_serial_nos = get_returned_serial_nos(source_doc, source_parent)
+		returned_batches = frappe._dict()
+		serial_and_batch_field = (
+			"serial_and_batch_bundle" if qty_field == "stock_qty" else "rejected_serial_and_batch_bundle"
+		)
+		old_serial_no_field = "serial_no" if qty_field == "stock_qty" else "rejected_serial_no"
+		old_batch_no_field = "batch_no"
 
-			type_of_transaction = "Inward"
-			if (
-				frappe.db.get_value(
-					"Serial and Batch Bundle", source_doc.serial_and_batch_bundle, "type_of_transaction"
-				)
-				== "Inward"
-			):
-				type_of_transaction = "Outward"
-
-			cls_obj = SerialBatchCreation(
-				{
-					"type_of_transaction": type_of_transaction,
-					"serial_and_batch_bundle": source_doc.serial_and_batch_bundle,
-					"returned_against": source_doc.name,
-					"item_code": source_doc.item_code,
-					"returned_serial_nos": returned_serial_nos,
-				}
-			)
-
-			cls_obj.duplicate_package()
-			if cls_obj.serial_and_batch_bundle:
-				target_doc.serial_and_batch_bundle = cls_obj.serial_and_batch_bundle
-
-		if source_doc.get("rejected_serial_and_batch_bundle"):
+		if (
+			source_doc.get(serial_and_batch_field)
+			or source_doc.get(old_serial_no_field)
+			or source_doc.get(old_batch_no_field)
+		):
 			if item_details.has_serial_no:
 				returned_serial_nos = get_returned_serial_nos(
-					source_doc, source_parent, serial_no_field="rejected_serial_and_batch_bundle"
+					source_doc, source_parent, serial_no_field=serial_and_batch_field
+				)
+			else:
+				returned_batches = get_returned_batches(
+					source_doc, source_parent, batch_no_field=serial_and_batch_field
 				)
 
 			type_of_transaction = "Inward"
-			if (
+			if source_doc.get(serial_and_batch_field) and (
 				frappe.db.get_value(
-					"Serial and Batch Bundle", source_doc.rejected_serial_and_batch_bundle, "type_of_transaction"
+					"Serial and Batch Bundle", source_doc.get(serial_and_batch_field), "type_of_transaction"
 				)
 				== "Inward"
 			):
 				type_of_transaction = "Outward"
+			elif source_parent.doctype in [
+				"Purchase Invoice",
+				"Purchase Receipt",
+				"Subcontracting Receipt",
+			]:
+				type_of_transaction = "Outward"
 
 			cls_obj = SerialBatchCreation(
 				{
 					"type_of_transaction": type_of_transaction,
-					"serial_and_batch_bundle": source_doc.rejected_serial_and_batch_bundle,
+					"serial_and_batch_bundle": source_doc.get(serial_and_batch_field),
 					"returned_against": source_doc.name,
 					"item_code": source_doc.item_code,
 					"returned_serial_nos": returned_serial_nos,
+					"voucher_type": source_parent.doctype,
+					"do_not_submit": True,
+					"warehouse": source_doc.warehouse,
+					"has_serial_no": item_details.has_serial_no,
+					"has_batch_no": item_details.has_batch_no,
 				}
 			)
 
-			cls_obj.duplicate_package()
-			if cls_obj.serial_and_batch_bundle:
-				target_doc.serial_and_batch_bundle = cls_obj.serial_and_batch_bundle
+			serial_nos = []
+			batches = frappe._dict()
+			if source_doc.get(old_batch_no_field):
+				batches = frappe._dict({source_doc.batch_no: source_doc.get(qty_field)})
+			elif source_doc.get(old_serial_no_field):
+				serial_nos = get_serial_nos(source_doc.get(old_serial_no_field))
+			elif source_doc.get(serial_and_batch_field):
+				if item_details.has_serial_no:
+					serial_nos = get_serial_nos_from_bundle(source_doc.get(serial_and_batch_field))
+				else:
+					batches = get_batches_from_bundle(source_doc.get(serial_and_batch_field))
 
+			if serial_nos:
+				cls_obj.serial_nos = sorted(list(set(serial_nos) - set(returned_serial_nos)))
+			elif batches:
+				for batch in batches:
+					if batch in returned_batches:
+						batches[batch] -= flt(returned_batches.get(batch))
+
+				cls_obj.batches = batches
+
+			if source_doc.get(serial_and_batch_field):
+				cls_obj.duplicate_package()
+				if cls_obj.serial_and_batch_bundle:
+					target_doc.set(serial_and_batch_field, cls_obj.serial_and_batch_bundle)
+			else:
+				target_doc.set(serial_and_batch_field, cls_obj.make_serial_and_batch_bundle().name)
+
+	def update_item(source_doc, target_doc, source_parent):
+		target_doc.qty = -1 * source_doc.qty
 		if doctype in ["Purchase Receipt", "Subcontracting Receipt"]:
 			returned_qty_map = get_returned_qty_map_for_row(
 				source_parent.name, source_parent.supplier, source_doc.name, doctype
@@ -561,6 +562,17 @@
 			if default_warehouse_for_sales_return:
 				target_doc.warehouse = default_warehouse_for_sales_return
 
+		item_details = frappe.get_cached_value(
+			"Item", source_doc.item_code, ["has_batch_no", "has_serial_no"], as_dict=1
+		)
+
+		if not item_details.has_batch_no and not item_details.has_serial_no:
+			return
+
+		for qty_field in ["stock_qty", "rejected_qty"]:
+			if target_doc.get(qty_field):
+				update_serial_batch_no(source_doc, target_doc, source_parent, item_details, qty_field)
+
 	def update_terms(source_doc, target_doc, source_parent):
 		target_doc.payment_amount = -source_doc.payment_amount
 
@@ -716,6 +728,9 @@
 		[parent_doc.doctype, "docstatus", "=", 1],
 	]
 
+	if serial_no_field == "rejected_serial_and_batch_bundle":
+		filters.append([child_doc.doctype, "rejected_qty", ">", 0])
+
 	# Required for POS Invoice
 	if ignore_voucher_detail_no:
 		filters.append([child_doc.doctype, "name", "!=", ignore_voucher_detail_no])
@@ -723,9 +738,57 @@
 	ids = []
 	for row in frappe.get_all(parent_doc.doctype, fields=fields, filters=filters):
 		ids.append(row.get("serial_and_batch_bundle"))
-		if row.get(old_field):
+		if row.get(old_field) and not row.get(serial_no_field):
 			serial_nos.extend(get_serial_nos_from_serial_no(row.get(old_field)))
 
-	serial_nos.extend(get_serial_nos(ids))
+	if ids:
+		serial_nos.extend(get_serial_nos(ids))
 
 	return serial_nos
+
+
+def get_returned_batches(
+	child_doc, parent_doc, batch_no_field=None, ignore_voucher_detail_no=None
+):
+	from erpnext.stock.serial_batch_bundle import get_batches_from_bundle
+
+	batches = frappe._dict()
+
+	old_field = "batch_no"
+	if not batch_no_field:
+		batch_no_field = "serial_and_batch_bundle"
+
+	return_ref_field = frappe.scrub(child_doc.doctype)
+	if child_doc.doctype == "Delivery Note Item":
+		return_ref_field = "dn_detail"
+
+	fields = [
+		f"`{'tab' + child_doc.doctype}`.`{batch_no_field}`",
+		f"`{'tab' + child_doc.doctype}`.`batch_no`",
+		f"`{'tab' + child_doc.doctype}`.`stock_qty`",
+	]
+
+	filters = [
+		[parent_doc.doctype, "return_against", "=", parent_doc.name],
+		[parent_doc.doctype, "is_return", "=", 1],
+		[child_doc.doctype, return_ref_field, "=", child_doc.name],
+		[parent_doc.doctype, "docstatus", "=", 1],
+	]
+
+	if batch_no_field == "rejected_serial_and_batch_bundle":
+		filters.append([child_doc.doctype, "rejected_qty", ">", 0])
+
+	# Required for POS Invoice
+	if ignore_voucher_detail_no:
+		filters.append([child_doc.doctype, "name", "!=", ignore_voucher_detail_no])
+
+	ids = []
+	for row in frappe.get_all(parent_doc.doctype, fields=fields, filters=filters):
+		ids.append(row.get("serial_and_batch_bundle"))
+		if row.get(old_field) and not row.get(batch_no_field):
+			batches.setdefault(row.get(old_field), row.get("stock_qty"))
+
+	if ids:
+		batches.update(get_batches_from_bundle(ids))
+
+	return batches
diff --git a/erpnext/controllers/selling_controller.py b/erpnext/controllers/selling_controller.py
index fdadb30..e8bae8c 100644
--- a/erpnext/controllers/selling_controller.py
+++ b/erpnext/controllers/selling_controller.py
@@ -308,6 +308,8 @@
 									"warehouse": p.warehouse or d.warehouse,
 									"item_code": p.item_code,
 									"qty": flt(p.qty),
+									"serial_no": p.serial_no if self.docstatus == 2 else None,
+									"batch_no": p.batch_no if self.docstatus == 2 else None,
 									"uom": p.uom,
 									"serial_and_batch_bundle": p.serial_and_batch_bundle
 									or get_serial_and_batch_bundle(p, self),
@@ -330,6 +332,8 @@
 							"warehouse": d.warehouse,
 							"item_code": d.item_code,
 							"qty": d.stock_qty,
+							"serial_no": d.serial_no if self.docstatus == 2 else None,
+							"batch_no": d.batch_no if self.docstatus == 2 else None,
 							"uom": d.uom,
 							"stock_uom": d.stock_uom,
 							"conversion_factor": d.conversion_factor,
diff --git a/erpnext/controllers/stock_controller.py b/erpnext/controllers/stock_controller.py
index 2a84204..671d2fb 100644
--- a/erpnext/controllers/stock_controller.py
+++ b/erpnext/controllers/stock_controller.py
@@ -463,6 +463,12 @@
 		sl_dict.update(args)
 		self.update_inventory_dimensions(d, sl_dict)
 
+		if self.docstatus == 2:
+			# To handle denormalized serial no records, will br deprecated in v16
+			for field in ["serial_no", "batch_no"]:
+				if d.get(field):
+					sl_dict[field] = d.get(field)
+
 		return sl_dict
 
 	def update_inventory_dimensions(self, row, sl_dict) -> None:
diff --git a/erpnext/public/js/controllers/buying.js b/erpnext/public/js/controllers/buying.js
index 3ed7fc7..77ecf75 100644
--- a/erpnext/public/js/controllers/buying.js
+++ b/erpnext/public/js/controllers/buying.js
@@ -361,9 +361,14 @@
 								new erpnext.SerialBatchPackageSelector(
 									me.frm, item, (r) => {
 										if (r) {
+											let qty = Math.abs(r.total_qty);
+											if (doc.is_return) {
+												qty = qty * -1;
+											}
+
 											let update_values = {
 												"serial_and_batch_bundle": r.name,
-												"qty": Math.abs(r.total_qty)
+												"qty": qty
 											}
 
 											if (r.warehouse) {
@@ -396,9 +401,14 @@
 								new erpnext.SerialBatchPackageSelector(
 									me.frm, item, (r) => {
 										if (r) {
+											let qty = Math.abs(r.total_qty);
+											if (doc.is_return) {
+												qty = qty * -1;
+											}
+
 											let update_values = {
 												"serial_and_batch_bundle": r.name,
-												"rejected_qty": Math.abs(r.total_qty)
+												"rejected_qty": qty
 											}
 
 											if (r.warehouse) {
diff --git a/erpnext/public/js/utils/sales_common.js b/erpnext/public/js/utils/sales_common.js
index 5514963..084cca7 100644
--- a/erpnext/public/js/utils/sales_common.js
+++ b/erpnext/public/js/utils/sales_common.js
@@ -317,9 +317,14 @@
 								new erpnext.SerialBatchPackageSelector(
 									me.frm, item, (r) => {
 										if (r) {
+											let qty = Math.abs(r.total_qty);
+											if (doc.is_return) {
+												qty = qty * -1;
+											}
+
 											frappe.model.set_value(item.doctype, item.name, {
 												"serial_and_batch_bundle": r.name,
-												"qty": Math.abs(r.total_qty)
+												"qty": qty
 											});
 										}
 									}
diff --git a/erpnext/public/js/utils/serial_no_batch_selector.js b/erpnext/public/js/utils/serial_no_batch_selector.js
index 7b9cdfe..4abc8fa 100644
--- a/erpnext/public/js/utils/serial_no_batch_selector.js
+++ b/erpnext/public/js/utils/serial_no_batch_selector.js
@@ -32,22 +32,39 @@
 		});
 
 		this.dialog.show();
-
-		let qty = this.item.stock_qty || this.item.transfer_qty || this.item.qty;
-		this.dialog.set_value("qty", qty).then(() => {
-			if (this.item.serial_no) {
-				this.dialog.set_value("scan_serial_no", this.item.serial_no);
-				frappe.model.set_value(this.item.doctype, this.item.name, 'serial_no', '');
-			} else if (this.item.batch_no) {
-				this.dialog.set_value("scan_batch_no", this.item.batch_no);
-				frappe.model.set_value(this.item.doctype, this.item.name, 'batch_no', '');
-			}
-
-			this.dialog.fields_dict.entries.grid.refresh();
-		});
-
 		this.$scan_btn = this.dialog.$wrapper.find(".link-btn");
 		this.$scan_btn.css("display", "inline");
+
+		let qty = this.item.stock_qty || this.item.transfer_qty || this.item.qty;
+
+		if (this.item?.is_rejected) {
+			qty = this.item.rejected_qty;
+		}
+
+		qty = Math.abs(qty);
+		if (qty > 0) {
+			this.dialog.set_value("qty", qty).then(() => {
+				if (this.item.serial_no && !this.item.serial_and_batch_bundle) {
+					let serial_nos = this.item.serial_no.split('\n');
+					if (serial_nos.length > 1) {
+						serial_nos.forEach(serial_no => {
+							this.dialog.fields_dict.entries.df.data.push({
+								serial_no: serial_no,
+								batch_no: this.item.batch_no
+							});
+						});
+					} else {
+						this.dialog.set_value("scan_serial_no", this.item.serial_no);
+					}
+					frappe.model.set_value(this.item.doctype, this.item.name, 'serial_no', '');
+				} else if (this.item.batch_no && !this.item.serial_and_batch_bundle) {
+					this.dialog.set_value("scan_batch_no", this.item.batch_no);
+					frappe.model.set_value(this.item.doctype, this.item.name, 'batch_no', '');
+				}
+
+				this.dialog.fields_dict.entries.grid.refresh();
+			});
+		}
 	}
 
 	get_serial_no_filters() {
@@ -467,13 +484,13 @@
 	}
 
 	render_data() {
-		if (!this.frm.is_new() && this.bundle) {
+		if (this.bundle) {
 			frappe.call({
 				method: 'erpnext.stock.doctype.serial_and_batch_bundle.serial_and_batch_bundle.get_serial_batch_ledgers',
 				args: {
 					item_code: this.item.item_code,
 					name: this.bundle,
-					voucher_no: this.item.parent,
+					voucher_no: !this.frm.is_new() ? this.item.parent : "",
 				}
 			}).then(r => {
 				if (r.message) {
diff --git a/erpnext/stock/doctype/delivery_note/test_delivery_note.py b/erpnext/stock/doctype/delivery_note/test_delivery_note.py
index 9465574..3a58122 100644
--- a/erpnext/stock/doctype/delivery_note/test_delivery_note.py
+++ b/erpnext/stock/doctype/delivery_note/test_delivery_note.py
@@ -174,6 +174,115 @@
 		for field, value in field_values.items():
 			self.assertEqual(cstr(serial_no.get(field)), value)
 
+	def test_delivery_note_return_against_denormalized_serial_no(self):
+		from erpnext.stock.doctype.delivery_note.delivery_note import make_sales_return
+		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
+
+		frappe.flags.ignore_serial_batch_bundle_validation = True
+		sn_item = "Old Serial NO Item Return Test - 1"
+		make_item(
+			sn_item,
+			{
+				"has_serial_no": 1,
+				"serial_no_series": "OSN-.####",
+				"is_stock_item": 1,
+			},
+		)
+
+		frappe.flags.ignore_serial_batch_bundle_validation = True
+		serial_nos = [
+			"OSN-1",
+			"OSN-2",
+			"OSN-3",
+			"OSN-4",
+			"OSN-5",
+			"OSN-6",
+			"OSN-7",
+			"OSN-8",
+			"OSN-9",
+			"OSN-10",
+			"OSN-11",
+			"OSN-12",
+		]
+
+		for sn in serial_nos:
+			if not frappe.db.exists("Serial No", sn):
+				sn_doc = frappe.get_doc(
+					{
+						"doctype": "Serial No",
+						"item_code": sn_item,
+						"serial_no": sn,
+					}
+				)
+				sn_doc.insert()
+
+		warehouse = "_Test Warehouse - _TC"
+		company = frappe.db.get_value("Warehouse", warehouse, "company")
+		se_doc = make_stock_entry(
+			item_code=sn_item,
+			company=company,
+			target="_Test Warehouse - _TC",
+			qty=12,
+			basic_rate=100,
+			do_not_submit=1,
+		)
+
+		se_doc.items[0].serial_no = "\n".join(serial_nos)
+		se_doc.submit()
+
+		self.assertEqual(sorted(get_serial_nos(se_doc.items[0].serial_no)), sorted(serial_nos))
+
+		dn = create_delivery_note(
+			item_code=sn_item,
+			qty=12,
+			rate=500,
+			warehouse=warehouse,
+			company=company,
+			expense_account="Cost of Goods Sold - _TC",
+			cost_center="Main - _TC",
+			do_not_submit=1,
+		)
+
+		dn.items[0].serial_no = "\n".join(serial_nos)
+		dn.submit()
+		dn.reload()
+
+		self.assertTrue(dn.items[0].serial_no)
+
+		frappe.flags.ignore_serial_batch_bundle_validation = False
+
+		# return entry
+		dn1 = make_sales_return(dn.name)
+
+		dn1.items[0].qty = -2
+
+		bundle_doc = frappe.get_doc("Serial and Batch Bundle", dn1.items[0].serial_and_batch_bundle)
+		bundle_doc.set("entries", bundle_doc.entries[:2])
+		bundle_doc.save()
+
+		dn1.save()
+		dn1.submit()
+
+		returned_serial_nos1 = get_serial_nos_from_bundle(dn1.items[0].serial_and_batch_bundle)
+		for serial_no in returned_serial_nos1:
+			self.assertTrue(serial_no in serial_nos)
+
+		dn2 = make_sales_return(dn.name)
+
+		dn2.items[0].qty = -2
+
+		bundle_doc = frappe.get_doc("Serial and Batch Bundle", dn2.items[0].serial_and_batch_bundle)
+		bundle_doc.set("entries", bundle_doc.entries[:2])
+		bundle_doc.save()
+
+		dn2.save()
+		dn2.submit()
+
+		returned_serial_nos2 = get_serial_nos_from_bundle(dn2.items[0].serial_and_batch_bundle)
+		for serial_no in returned_serial_nos2:
+			self.assertTrue(serial_no in serial_nos)
+			self.assertFalse(serial_no in returned_serial_nos1)
+
 	def test_sales_return_for_non_bundled_items_partial(self):
 		company = frappe.db.get_value("Warehouse", "Stores - TCP1", "company")
 
diff --git a/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py b/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
index ecb9314..52ef26e 100644
--- a/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
+++ b/erpnext/stock/doctype/serial_and_batch_bundle/serial_and_batch_bundle.py
@@ -23,7 +23,11 @@
 )
 from frappe.utils.csvutils import build_csv_response
 
-from erpnext.stock.serial_batch_bundle import BatchNoValuation, SerialNoValuation
+from erpnext.stock.serial_batch_bundle import (
+	BatchNoValuation,
+	SerialNoValuation,
+	get_batches_from_bundle,
+)
 from erpnext.stock.serial_batch_bundle import get_serial_nos as get_serial_nos_from_bundle
 
 
@@ -123,6 +127,11 @@
 				)
 
 	def validate_serial_nos_duplicate(self):
+		# Don't inward same serial number multiple times
+
+		if not self.warehouse:
+			return
+
 		if self.voucher_type in ["Stock Reconciliation", "Stock Entry"] and self.docstatus != 1:
 			return
 
@@ -146,7 +155,6 @@
 			kwargs["voucher_no"] = self.voucher_no
 
 		available_serial_nos = get_available_serial_nos(kwargs)
-
 		for data in available_serial_nos:
 			if data.serial_no in serial_nos:
 				self.throw_error_message(
@@ -327,6 +335,19 @@
 		):
 			values_to_set["posting_time"] = parent.posting_time
 
+		if parent.doctype in [
+			"Delivery Note",
+			"Purchase Receipt",
+			"Purchase Invoice",
+			"Sales Invoice",
+		] and parent.get("is_return"):
+			return_ref_field = frappe.scrub(parent.doctype) + "_item"
+			if parent.doctype == "Delivery Note":
+				return_ref_field = "dn_detail"
+
+			if row.get(return_ref_field):
+				values_to_set["returned_against"] = row.get(return_ref_field)
+
 		if values_to_set:
 			self.db_set(values_to_set)
 
@@ -509,7 +530,6 @@
 		batch_nos = []
 
 		serial_batches = {}
-
 		for row in self.entries:
 			if self.has_serial_no and not row.serial_no:
 				frappe.throw(
@@ -590,6 +610,67 @@
 				f"Batch Nos {bold(incorrect_batch_nos)} does not belong to Item {bold(self.item_code)}"
 			)
 
+	def validate_serial_and_batch_no_for_returned(self):
+		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos
+
+		if not self.returned_against:
+			return
+
+		if self.voucher_type not in [
+			"Purchase Receipt",
+			"Purchase Invoice",
+			"Sales Invoice",
+			"Delivery Note",
+		]:
+			return
+
+		data = self.get_orignal_document_data()
+		if not data:
+			return
+
+		serial_nos, batches = [], []
+		current_serial_nos = [d.serial_no for d in self.entries if d.serial_no]
+		current_batches = [d.batch_no for d in self.entries if d.batch_no]
+
+		for d in data:
+			if self.has_serial_no:
+				if d.serial_and_batch_bundle:
+					serial_nos = get_serial_nos_from_bundle(d.serial_and_batch_bundle)
+				else:
+					serial_nos = get_serial_nos(d.serial_no)
+
+			elif self.has_batch_no:
+				if d.serial_and_batch_bundle:
+					batches = get_batches_from_bundle(d.serial_and_batch_bundle)
+				else:
+					batches = frappe._dict({d.batch_no: d.stock_qty})
+
+				if batches:
+					batches = [d for d in batches if batches[d] > 0]
+
+			if serial_nos:
+				if not set(current_serial_nos).issubset(set(serial_nos)):
+					self.throw_error_message(
+						f"Serial Nos {bold(', '.join(serial_nos))} are not part of the original document."
+					)
+
+			if batches:
+				if not set(current_batches).issubset(set(batches)):
+					self.throw_error_message(
+						f"Batch Nos {bold(', '.join(batches))} are not part of the original document."
+					)
+
+	def get_orignal_document_data(self):
+		fields = ["serial_and_batch_bundle", "stock_qty"]
+		if self.has_serial_no:
+			fields.append("serial_no")
+
+		elif self.has_batch_no:
+			fields.append("batch_no")
+
+		child_doc = self.voucher_type + " Item"
+		return frappe.get_all(child_doc, fields=fields, filters={"name": self.returned_against})
+
 	def validate_duplicate_serial_and_batch_no(self):
 		serial_nos = []
 		batch_nos = []
@@ -688,9 +769,29 @@
 		for batch in batches:
 			frappe.db.set_value("Batch", batch.name, {"reference_name": None, "reference_doctype": None})
 
+	def before_submit(self):
+		self.validate_serial_and_batch_no_for_returned()
+		self.set_purchase_document_no()
+
 	def on_submit(self):
 		self.validate_serial_nos_inventory()
 
+	def set_purchase_document_no(self):
+		if not self.has_serial_no:
+			return
+
+		if self.total_qty > 0:
+			serial_nos = [d.serial_no for d in self.entries if d.serial_no]
+			sn_table = frappe.qb.DocType("Serial No")
+			(
+				frappe.qb.update(sn_table)
+				.set(
+					sn_table.purchase_document_no,
+					self.voucher_no if not sn_table.purchase_document_no else self.voucher_no,
+				)
+				.where(sn_table.name.isin(serial_nos))
+			).run()
+
 	def validate_serial_and_batch_inventory(self):
 		self.check_future_entries_exists()
 		self.validate_batch_inventory()
diff --git a/erpnext/stock/doctype/serial_no/serial_no.json b/erpnext/stock/doctype/serial_no/serial_no.json
index b4ece00..2d7fcac 100644
--- a/erpnext/stock/doctype/serial_no/serial_no.json
+++ b/erpnext/stock/doctype/serial_no/serial_no.json
@@ -27,8 +27,6 @@
   "column_break_24",
   "location",
   "employee",
-  "delivery_details",
-  "delivery_document_type",
   "warranty_amc_details",
   "column_break6",
   "warranty_expiry_date",
@@ -39,7 +37,8 @@
   "more_info",
   "company",
   "column_break_2cmm",
-  "work_order"
+  "work_order",
+  "purchase_document_no"
  ],
  "fields": [
   {
@@ -154,20 +153,6 @@
    "read_only": 1
   },
   {
-   "fieldname": "delivery_details",
-   "fieldtype": "Section Break",
-   "label": "Delivery Details",
-   "oldfieldtype": "Column Break"
-  },
-  {
-   "fieldname": "delivery_document_type",
-   "fieldtype": "Link",
-   "label": "Delivery Document Type",
-   "no_copy": 1,
-   "options": "DocType",
-   "read_only": 1
-  },
-  {
    "fieldname": "warranty_amc_details",
    "fieldtype": "Section Break",
    "label": "Warranty / AMC Details"
@@ -275,12 +260,19 @@
   {
    "fieldname": "column_break_2cmm",
    "fieldtype": "Column Break"
+  },
+  {
+   "fieldname": "purchase_document_no",
+   "fieldtype": "Data",
+   "label": "Creation Document No",
+   "no_copy": 1,
+   "read_only": 1
   }
  ],
  "icon": "fa fa-barcode",
  "idx": 1,
  "links": [],
- "modified": "2023-11-28 15:37:59.489945",
+ "modified": "2023-12-17 10:52:55.767839",
  "modified_by": "Administrator",
  "module": "Stock",
  "name": "Serial No",
diff --git a/erpnext/stock/doctype/serial_no/serial_no.py b/erpnext/stock/doctype/serial_no/serial_no.py
index d562560..122664c 100644
--- a/erpnext/stock/doctype/serial_no/serial_no.py
+++ b/erpnext/stock/doctype/serial_no/serial_no.py
@@ -41,7 +41,6 @@
 		batch_no: DF.Link | None
 		brand: DF.Link | None
 		company: DF.Link
-		delivery_document_type: DF.Link | None
 		description: DF.Text | None
 		employee: DF.Link | None
 		item_code: DF.Link
@@ -51,6 +50,7 @@
 		maintenance_status: DF.Literal[
 			"", "Under Warranty", "Out of Warranty", "Under AMC", "Out of AMC"
 		]
+		purchase_document_no: DF.Data | None
 		purchase_rate: DF.Float
 		serial_no: DF.Data
 		status: DF.Literal["", "Active", "Inactive", "Delivered", "Expired"]
@@ -231,26 +231,6 @@
 	return sorted([d.get("name") for d in serial_numbers])
 
 
-def get_delivered_serial_nos(serial_nos):
-	"""
-	Returns serial numbers that delivered from the list of serial numbers
-	"""
-	from frappe.query_builder.functions import Coalesce
-
-	SerialNo = frappe.qb.DocType("Serial No")
-	serial_nos = get_serial_nos(serial_nos)
-	query = (
-		frappe.qb.select(SerialNo.name)
-		.from_(SerialNo)
-		.where((SerialNo.name.isin(serial_nos)) & (Coalesce(SerialNo.delivery_document_type, "") != ""))
-	)
-
-	result = query.run()
-	if result and len(result) > 0:
-		delivered_serial_nos = [row[0] for row in result]
-		return delivered_serial_nos
-
-
 @frappe.whitelist()
 def get_pos_reserved_serial_nos(filters):
 	if isinstance(filters, str):
diff --git a/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py b/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py
index 23788cf..6e7af68 100644
--- a/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py
+++ b/erpnext/stock/doctype/stock_ledger_entry/stock_ledger_entry.py
@@ -181,6 +181,9 @@
 			frappe.throw(_("Actual Qty is mandatory"))
 
 	def validate_serial_batch_no_bundle(self):
+		if self.is_cancelled == 1:
+			return
+
 		item_detail = frappe.get_cached_value(
 			"Item",
 			self.item_code,
diff --git a/erpnext/stock/serial_batch_bundle.py b/erpnext/stock/serial_batch_bundle.py
index 0c18792..a1874b8 100644
--- a/erpnext/stock/serial_batch_bundle.py
+++ b/erpnext/stock/serial_batch_bundle.py
@@ -218,15 +218,16 @@
 			).validate_serial_and_batch_inventory()
 
 	def post_process(self):
-		if not self.sle.serial_and_batch_bundle:
+		if not self.sle.serial_and_batch_bundle and not self.sle.serial_no and not self.sle.batch_no:
 			return
 
-		docstatus = frappe.get_cached_value(
-			"Serial and Batch Bundle", self.sle.serial_and_batch_bundle, "docstatus"
-		)
+		if self.sle.serial_and_batch_bundle:
+			docstatus = frappe.get_cached_value(
+				"Serial and Batch Bundle", self.sle.serial_and_batch_bundle, "docstatus"
+			)
 
-		if docstatus != 1:
-			self.submit_serial_and_batch_bundle()
+			if docstatus != 1:
+				self.submit_serial_and_batch_bundle()
 
 		if self.item_details.has_serial_no == 1:
 			self.set_warehouse_and_status_in_serial_nos()
@@ -249,7 +250,12 @@
 		doc.submit()
 
 	def set_warehouse_and_status_in_serial_nos(self):
+		from erpnext.stock.doctype.serial_no.serial_no import get_serial_nos as get_parsed_serial_nos
+
 		serial_nos = get_serial_nos(self.sle.serial_and_batch_bundle)
+		if not self.sle.serial_and_batch_bundle and self.sle.serial_no:
+			serial_nos = get_parsed_serial_nos(self.sle.serial_no)
+
 		warehouse = self.warehouse if self.sle.actual_qty > 0 else None
 
 		if not serial_nos:
@@ -263,7 +269,14 @@
 		(
 			frappe.qb.update(sn_table)
 			.set(sn_table.warehouse, warehouse)
-			.set(sn_table.status, "Active" if warehouse else status)
+			.set(
+				sn_table.status,
+				"Active"
+				if warehouse
+				else status
+				if (sn_table.purchase_document_no != self.sle.voucher_no and self.sle.is_cancelled != 1)
+				else "Inactive",
+			)
 			.where(sn_table.name.isin(serial_nos))
 		).run()
 
@@ -290,6 +303,8 @@
 		from erpnext.stock.doctype.batch.batch import get_available_batches
 
 		batches = get_batch_nos(self.sle.serial_and_batch_bundle)
+		if not self.sle.serial_and_batch_bundle and self.sle.batch_no:
+			batches = frappe._dict({self.sle.batch_no: self.sle.actual_qty})
 
 		batches_qty = get_available_batches(
 			frappe._dict(
@@ -312,13 +327,35 @@
 	if serial_nos:
 		filters["serial_no"] = ("in", serial_nos)
 
-	entries = frappe.get_all("Serial and Batch Entry", fields=["serial_no"], filters=filters)
+	entries = frappe.get_all(
+		"Serial and Batch Entry", fields=["serial_no"], filters=filters, order_by="idx"
+	)
 	if not entries:
 		return []
 
 	return [d.serial_no for d in entries if d.serial_no]
 
 
+def get_batches_from_bundle(serial_and_batch_bundle, batches=None):
+	if not serial_and_batch_bundle:
+		return []
+
+	filters = {"parent": serial_and_batch_bundle, "batch_no": ("is", "set")}
+	if isinstance(serial_and_batch_bundle, list):
+		filters = {"parent": ("in", serial_and_batch_bundle)}
+
+	if batches:
+		filters["batch_no"] = ("in", batches)
+
+	entries = frappe.get_all(
+		"Serial and Batch Entry", fields=["batch_no", "qty"], filters=filters, order_by="idx", as_list=1
+	)
+	if not entries:
+		return frappe._dict({})
+
+	return frappe._dict(entries)
+
+
 def get_serial_nos_from_bundle(serial_and_batch_bundle, serial_nos=None):
 	return get_serial_nos(serial_and_batch_bundle, serial_nos=serial_nos)