Credit and bound memory from aborted vamana inserts

sql/svs--0.1.0.sql

@@ -162,6 +162,7 @@ CREATE FUNCTION pg_stat_vamana_worker()
 		evict_all                       bool,
 		heartbeat_ts                    timestamptz,
 		residency_bytes_committed       bigint,
+		residency_bytes_reclaimable     bigint,
 		build_bytes_committed           bigint,
 		residency_memory_limit          bigint,
 		search_work_mem_limit           bigint,
@@ -185,6 +186,7 @@ CREATE VIEW pg_stat_vamana_worker AS
 		   w.evict_all,
 		   w.heartbeat_ts,
 		   w.residency_bytes_committed,
+		   w.residency_bytes_reclaimable,
 		   w.build_bytes_committed,
 		   w.residency_memory_limit,
 		   w.residency_bytes_committed - COALESCE(r.resident_bytes, 0) AS residency_drift,

src/svs_memory.c

@@ -221,6 +221,25 @@ FreeReservation(SvsMemReservation *reservation)
 	reservation->relid = InvalidOid;
 }
 
+/*
+ * Call under entry->memLock (shared or exclusive). Total real memory this
+ * database holds beyond residencyBytesCommitted, from aborted inserts not
+ * yet reclaimed by a compaction. A plain scan, not a second counter to keep
+ * in step: VAMANA_MAX_INDEXES is small and this is only read on the
+ * abort/credit path and the stats view.
+ */
+static uint64
+SumReclaimableBytes(VamanaWorkerShmem *entry)
+{
+	uint64		total = 0;
+
+	for (int i = 0; i < VAMANA_MAX_INDEXES; i++)
+		if (entry->reservations[i].relid != InvalidOid)
+			total += entry->reservations[i].reclaimableBytes;
+
+	return total;
+}
+
 /*
  * Returns a REBUILDING reservation to RESIDENT at exactly the size it held
  * before the rebuild began, on every path off of REBUILDING other than a
@@ -382,7 +401,8 @@ SvsMemoryAdmitDatabase(Oid dbOid, uint64 residencyBudget, uint64 durableCommitte
 	LWLockAcquire(&entry->memLock, LW_EXCLUSIVE);
 
 	{
-		uint64		committedFloor = Max(entry->residencyBytesCommitted, durableCommittedFloor);
+		uint64		realUsage = entry->residencyBytesCommitted + SumReclaimableBytes(entry);
+		uint64		committedFloor = Max(realUsage, durableCommittedFloor);
 
 		if (residencyBudget < committedFloor)
 		{
@@ -437,7 +457,8 @@ SvsMemoryRestoreResidencyBudget(Oid dbOid, uint64 priorBudget)
 
 	LWLockAcquire(&entry->memLock, LW_EXCLUSIVE);
 
-	restoredBudget = Max(priorBudget, entry->residencyBytesCommitted);
+	restoredBudget = Max(priorBudget,
+						  entry->residencyBytesCommitted + SumReclaimableBytes(entry));
 	if (restoredBudget != priorBudget)
 		ereport(WARNING,
 				(errmsg("SVS memory accounting: database %u's residency budget restored to %llu bytes on transaction abort, not its pre-transaction %llu, to cover bytes already committed under the aborted value",
@@ -651,6 +672,14 @@ SvsMemoryConfirmBuild(Oid dbOid, Oid relid, uint64 buildPeak, uint64 measuredRes
 		reservation->state = SVS_MEM_CONFIRMED;
 		reservation->measuredBytes = measuredResidencyBytes;
 		reservation->priorResidentBytes = 0;
+
+		/*
+		 * A freshly confirmed build or rebuild is a newly measured graph
+		 * with no aborted-insert history of its own; any debt the old
+		 * graph was carrying (isRebuild) does not apply to it.
+		 */
+		reservation->reclaimableBytes = 0;
+		reservation->residentGeneration = ++entry->nextResidentGeneration;
 	}
 	else if (isRebuild)
 	{
@@ -820,6 +849,10 @@ SvsMemoryReconcileLoad(Oid dbOid, Oid relid, uint64 measuredBytes, uint64 capaci
 			reservation->capacityHeadroomVectors = capacityHeadroomVectors;
 			reservation->priorResidentBytes = 0;
 			reservation->buildPeakBytes = 0;
+
+			/* A freshly loaded graph starts with no aborted-insert debt. */
+			reservation->reclaimableBytes = 0;
+			reservation->residentGeneration = ++entry->nextResidentGeneration;
 		}
 	}
 	else
@@ -835,6 +868,7 @@ SvsMemoryReconcileLoad(Oid dbOid, Oid relid, uint64 measuredBytes, uint64 capaci
 			reservation->estimateBytes = measuredBytes;
 			reservation->measuredBytes = measuredBytes;
 			reservation->capacityHeadroomVectors = capacityHeadroomVectors;
+			reservation->residentGeneration = ++entry->nextResidentGeneration;
 
 			entry->residencyBytesCommitted += measuredBytes;
 		}
@@ -971,31 +1005,58 @@ WarnIfResidencyOverBudget(VamanaWorkerShmem *entry, Oid dbOid, Oid relid, const
 }
 
 void
-SvsMemoryReanchorInsert(Oid dbOid, Oid relid, uint64 measuredBytes)
+SvsMemoryReanchorInsert(Oid dbOid, Oid relid, uint64 measuredBytes,
+						 uint64 *growthBytesOut, uint32 *generationOut)
 {
 	VamanaWorkerShmem *entry = LookupEntryOrError(dbOid);
 	SvsMemReservation *reservation;
+	uint64		oldRaw;
+	uint64		growth;
+	uint64		newNet;
 
 	Assert(OidIsValid(relid));
 
 	LWLockAcquire(&entry->memLock, LW_EXCLUSIVE);
 
 	reservation = FindResidentReservationOrError(entry, dbOid, relid, "reanchor");
 
+	/*
+	 * oldRaw is the last raw measurement this reservation stood for (net
+	 * figure plus whatever was already reclaimable); growth is however much
+	 * larger the fresh raw measurement is than that. A delete-driven drop
+	 * (SVS frees only a little before compaction, but not never) clamps the
+	 * debt down to the new raw total rather than going negative; it never
+	 * grows the debt here -- only SvsMemoryCreditAbortedInserts does that.
+	 */
+	oldRaw = reservation->measuredBytes + reservation->reclaimableBytes;
+	growth = (measuredBytes > oldRaw) ? (measuredBytes - oldRaw) : 0;
+
+	if (measuredBytes < reservation->reclaimableBytes)
+		reservation->reclaimableBytes = measuredBytes;
+
+	newNet = measuredBytes - reservation->reclaimableBytes;
+
 	SubtractFloored(&entry->residencyBytesCommitted, reservation->measuredBytes,
 					"an index's pre-reanchor residency");
 	ReleaseOldestInsertReservation(entry, relid);
-	entry->residencyBytesCommitted += measuredBytes;
-	reservation->measuredBytes = measuredBytes;
+	entry->residencyBytesCommitted += newNet;
+	reservation->measuredBytes = newNet;
+
+	Assert(reservation->measuredBytes + reservation->reclaimableBytes == measuredBytes);
 
 	WarnIfResidencyOverBudget(entry, dbOid, relid, "an insert into");
 
+	if (growthBytesOut != NULL)
+		*growthBytesOut = growth;
+	if (generationOut != NULL)
+		*generationOut = reservation->residentGeneration;
+
 	LWLockRelease(&entry->memLock);
 }
 
 void
 SvsMemoryReconcileResident(Oid dbOid, Oid relid, uint64 measuredBytes,
-						   uint64 capacityHeadroomVectors)
+						   uint64 *capacityHeadroomVectors)
 {
 	VamanaWorkerShmem *entry = LookupEntryOrError(dbOid);
 	SvsMemReservation *reservation;
@@ -1018,7 +1079,15 @@ SvsMemoryReconcileResident(Oid dbOid, Oid relid, uint64 measuredBytes,
 					"an index's pre-reconcile residency");
 	entry->residencyBytesCommitted += measuredBytes;
 	reservation->measuredBytes = measuredBytes;
-	reservation->capacityHeadroomVectors = capacityHeadroomVectors;
+	if (capacityHeadroomVectors != NULL)
+		reservation->capacityHeadroomVectors = *capacityHeadroomVectors;
+
+	/*
+	 * measuredBytes here is always a post-compaction raw measurement (see
+	 * the contract on this function in svs_memory.h): every byte SVS still
+	 * holds is counted above, net, so nothing is left to call reclaimable.
+	 */
+	reservation->reclaimableBytes = 0;
 
 	WarnIfResidencyOverBudget(entry, dbOid, relid, "a compact of");
 
@@ -1071,6 +1140,45 @@ SvsMemoryAbortInsert(Oid dbOid, Oid relid)
 	LWLockRelease(&entry->memLock);
 }
 
+bool
+SvsMemoryCreditAbortedInserts(Oid dbOid, Oid relid, uint32 generation, uint64 growthBytes)
+{
+	VamanaWorkerShmem *entry = VamanaWorkerLookupSlot(dbOid);
+	SvsMemReservation *reservation;
+	bool		overCap = false;
+
+	Assert(OidIsValid(relid));
+
+	if (entry == NULL)
+		return false;
+
+	LWLockAcquire(&entry->memLock, LW_EXCLUSIVE);
+
+	reservation = FindReservation(entry, relid);
+
+	if (reservation != NULL &&
+		reservation->state == SVS_MEM_RESIDENT &&
+		reservation->residentGeneration == generation &&
+		vamana_compact_threshold_pct < 100)
+	{
+		uint64		credit = Min(growthBytes, reservation->measuredBytes);
+		uint64		cap;
+
+		SubtractFloored(&entry->residencyBytesCommitted, credit,
+						 "an aborted insert's credited growth");
+		SubtractFloored(&reservation->measuredBytes, credit,
+						 "an aborted insert's reservation contribution");
+		reservation->reclaimableBytes += credit;
+
+		cap = (entry->residencyBudget * (uint64) vamana_compact_threshold_pct) / 100;
+		overCap = SumReclaimableBytes(entry) > cap;
+	}
+
+	LWLockRelease(&entry->memLock);
+
+	return overCap;
+}
+
 /*
  * A live PID can belong to a recycled process, not the original owner --
  * cross-check reservedAt against the current occupant's start time. An
@@ -1288,6 +1396,7 @@ SvsMemoryReadStats(Oid dbOid, SvsMemoryStats *out)
 	out->residencyBudget = entry->residencyBudget;
 	out->residencyBytesCommitted = entry->residencyBytesCommitted;
 	out->buildBytesCommitted = entry->buildBytesCommitted;
+	out->residencyBytesReclaimable = SumReclaimableBytes(entry);
 
 	LWLockRelease(&entry->memLock);
 
@@ -1453,6 +1562,9 @@ SvsMemoryResetDatabaseAccounting(VamanaWorkerShmem *entry)
 	entry->residencyBudget = 0;
 	entry->residencyBytesCommitted = 0;
 	entry->buildBytesCommitted = 0;
+	/* nextResidentGeneration is deliberately not reset here: a released
+	 * slot's next occupant must not collide with a generation some
+	 * backend's in-flight undo entry still cites. */
 	pg_atomic_write_u64(&entry->searchScratchBytesInFlight, 0);
 
 	for (int i = 0; i < VAMANA_MAX_INDEXES; i++)

src/svs_memory.h

@@ -57,6 +57,16 @@ extern PGDLLIMPORT int vamana_default_residency_memory_mb;
 extern PGDLLIMPORT int vamana_max_search_work_mem_mb;
 extern PGDLLIMPORT int vamana_default_search_work_mem_mb;
 
+/*
+ * Owned and registered by vamana.c; declared here too (rather than pulling
+ * in the much heavier vamana.h, which test/modules/svs_memory_test compiles
+ * this module without) because SvsMemoryCreditAbortedInserts reads it
+ * directly: svs.compact_threshold_pct doubles as this module's "how much
+ * reclaimable debt to carry before forcing a compact" cap, not just
+ * VACUUM's own compact trigger.
+ */
+extern PGDLLIMPORT int vamana_compact_threshold_pct;
+
 typedef enum SvsMemReservationState
 {
 	SVS_MEM_RESERVED,
@@ -126,6 +136,35 @@ typedef struct SvsMemReservation
 	 * insert never applies.
 	 */
 	uint64		capacityHeadroomVectors;
+
+	/*
+	 * Bytes contributed by inserts that were later rolled back, still
+	 * physically resident in the SVS graph until the next compaction (SVS
+	 * never reuses a deleted row's slot). Meaningful only in RESIDENT and
+	 * REBUILDING; zero otherwise. The invariant this field exists to hold:
+	 *
+	 *     measuredBytes + reclaimableBytes == the last raw SVSGetIndexMemoryUsage()
+	 *     taken for this reservation.
+	 *
+	 * measuredBytes is therefore the net figure the residency budget gate
+	 * reads; reclaimableBytes is the bounded, visible gap between that net
+	 * figure and real SVS memory. Zeroed whenever a fresh measurement
+	 * already reflects a compaction: SvsMemoryReconcileResident,
+	 * SvsMemoryReconcileLoad, and a successful SvsMemoryConfirmBuild.
+	 */
+	uint64		reclaimableBytes;
+
+	/*
+	 * Monotonic stamp of which measured graph this reservation currently
+	 * holds, assigned from the database's nextResidentGeneration counter by
+	 * SvsMemoryReconcileLoad and SvsMemoryConfirmBuild -- the two places a
+	 * fresh graph is measured -- and never by a compaction. Lets
+	 * SvsMemoryCreditAbortedInserts tell "this abort's growth belongs to the
+	 * graph still resident" from "the graph was reloaded or rebuilt since,
+	 * and the undo entry's growth figure no longer means anything," so a
+	 * stale credit is skipped rather than applied to the wrong graph.
+	 */
+	uint32		residentGeneration;
 } SvsMemReservation;
 
 /*
@@ -294,16 +333,68 @@ extern bool SvsMemoryReserveInsert(Oid dbOid, Oid relid, uint64 deltaBytes);
 /*
  * Worker, after applying an insert batch under the index's write lock.
  * Folds the oldest pending insert reservation for relid, plus relid's prior
- * committed size, into the single fresh exact measurement measuredBytes.
+ * committed size, into the single fresh exact measurement measuredBytes,
+ * net of whatever this reservation's reclaimableBytes debt already covers
+ * (see SvsMemReservation's invariant). *growthBytesOut receives how much of
+ * measuredBytes, relative to the reservation's last raw measurement, this
+ * particular reanchor added -- 0 if the index shrank or stayed the same.
+ * *generationOut receives the reservation's current residentGeneration.
+ * The caller carries both through to the undo log so a later abort of this
+ * same insert can credit exactly this growth back as reclaimableBytes, via
+ * SvsMemoryCreditAbortedInserts. Either output pointer may be NULL.
  */
-extern void SvsMemoryReanchorInsert(Oid dbOid, Oid relid, uint64 measuredBytes);
+extern void SvsMemoryReanchorInsert(Oid dbOid, Oid relid, uint64 measuredBytes,
+									 uint64 *growthBytesOut, uint32 *generationOut);
 
 /*
  * Like SvsMemoryReanchorInsert, but never touches pending insert
  * reservations for relid -- those belong to unrelated, unapplied inserts.
+ * Contract: measuredBytes must already be a raw measurement taken after a
+ * compaction (the COMPACT maintenance path, or a checkpoint's save, both of
+ * which compact the live graph before this is called), never a delete- or
+ * insert-driven re-measurement. This zeroes the reservation's
+ * reclaimableBytes to match: once a compaction has run, every byte SVS
+ * still holds is counted, net, with nothing left to call reclaimable.
+ *
+ * capacityHeadroomVectors, if not NULL, replaces the reservation's cached
+ * figure. Pass NULL when the caller cannot safely recompute it (e.g. a
+ * checkpoint's error path, where re-reading the metapage to refresh
+ * headroom could re-lock a buffer the failure left exclusively locked);
+ * compaction alone never changes vector count, so the existing figure is
+ * still correct in that case.
  */
 extern void SvsMemoryReconcileResident(Oid dbOid, Oid relid, uint64 measuredBytes,
-										uint64 capacityHeadroomVectors);
+										uint64 *capacityHeadroomVectors);
+
+/*
+ * Worker or backend abort path, once an aborted insert's rows are confirmed
+ * deleted from the graph (VamanaWorkerSubmitDelete returned true for the
+ * undo batch that covered them). Moves min(growthBytes,
+ * reservation->measuredBytes) from the net committed figure into
+ * reclaimableBytes, so the aborted bytes stop counting against the budget
+ * without claiming memory that was never really freed: SVS does not reuse a
+ * deleted row's slot before a compaction, so the bytes stay physically
+ * resident until one runs.
+ *
+ * generation must match the reservation's current residentGeneration or
+ * this is a no-op: the graph has been reloaded or rebuilt since the insert
+ * applied, and growthBytes no longer describes anything live in it (see
+ * SvsMemReservation.residentGeneration). Also a no-op if there is no slot,
+ * no reservation, the reservation is not RESIDENT, or
+ * svs.compact_threshold_pct is 100 (the operator's existing way to say
+ * "never compact," which this treats as "never refund either" -- no
+ * reclaim backstop means no uncompacted debt is safe to carry). Never
+ * errors, the same contract SvsMemoryRestoreResidencyBudget follows for a
+ * call reached from an abort path.
+ *
+ * Returns true if, after crediting, dbOid's total reclaimable bytes across
+ * every reservation exceed svs.compact_threshold_pct of its residency
+ * budget -- the caller's signal to request a COMPACT for relid once its
+ * batch finishes, which is the only thing that ever reduces this total back
+ * down.
+ */
+extern bool SvsMemoryCreditAbortedInserts(Oid dbOid, Oid relid, uint32 generation,
+										   uint64 growthBytes);
 
 /*
  * Worker, on the empty-table first-insert build path: relid's reservation
@@ -351,6 +442,14 @@ typedef struct SvsMemoryStats
 	uint64		residencyBudget;
 	uint64		residencyBytesCommitted;
 	uint64		buildBytesCommitted;
+
+	/*
+	 * Sum of every reservation's reclaimableBytes: real SVS memory this
+	 * database holds beyond residencyBytesCommitted, from aborted inserts
+	 * not yet reclaimed by a compaction. Bounded by svs.compact_threshold_pct
+	 * of residencyBudget; see SvsMemoryCreditAbortedInserts.
+	 */
+	uint64		residencyBytesReclaimable;
 } SvsMemoryStats;
 
 /*

src/vamana.c

@@ -249,7 +249,12 @@ VamanaInit(void)
 							"0 = compact on every VACUUM with pending deletes. "
 							"100 = disable compact (consolidate still runs). "
 							"Higher values reduce compact frequency and memory reclamation; "
-							"lower values keep the index tighter at the cost of more frequent compacts.",
+							"lower values keep the index tighter at the cost of more frequent compacts. "
+							"Also governs a second, independent threshold: a rolled-back INSERT's "
+							"bytes are credited back as reclaimable residency rather than left "
+							"stranded against the budget, up to this percentage of a database's "
+							"residency budget before a compact is forced to reclaim them; 100 "
+							"disables that credit the same way it disables VACUUM's own compacts.",
 							&vamana_compact_threshold_pct,
 							10, 0, 100,
 							PGC_USERSET,