Review comments, eviction and minor simplification

Author: dbeck

src/chunk.c

@@ -40,6 +40,7 @@
 #include <storage/lmgr.h>
 #include <storage/lockdefs.h>
 #include <tcop/tcopprot.h>
+#include <ts_stats/ts_stats_record.h>
 #include <utils/acl.h>
 #include <utils/array.h>
 #include <utils/builtins.h>
@@ -4074,6 +4075,9 @@ ts_chunk_drop(const Chunk *chunk, DropBehavior behavior, int32 log_level)
 
 	/* Drop the table */
 	performDeletion(&objaddr, behavior, 0);
+
+	/* Evict the chunk stats from the shared memory */
+	ts_stats_chunk_evict(chunk->table_id);
 }
 
 static void

src/guc.c

@@ -1489,8 +1489,7 @@ _guc_init(void)
 							"Per-database statistics cache capacity, "
 							"in chunks. 0 disables the feature.",
 							"Must be 0 or a power of 2. "
-							"Takes effect only for databases whose observability segment "
-							"has not yet been created.",
+							"Takes effect only after server restart.",
 							&ts_guc_stats_max_chunks,
 							TS_STATS_MAX_CHUNKS_DEFAULT, /* default */
 							0,							 /* min: 0 = disabled */

src/ts_stats/ts_stats_defs.h

@@ -5,8 +5,10 @@
  */
 #pragma once
 
+#include <string.h>
 #include <postgres.h>
 #include "export.h"
+#include <c.h>
 #include <chunk_insert_state.h>
 #include <datatype/timestamp.h>
 #include <port/atomics.h>
@@ -54,6 +56,12 @@ StaticAssertDecl(TS_STATS_BUCKETS < 256 && TS_STATS_PROBE_WIDTH < 256,
  * the order of recency.
  */
 
+typedef struct TsStatsRelids
+{
+	Oid compressed_relid;
+	Oid uncompressed_relid;
+} TsStatsRelids;
+
 /*
  * As explained above, TsStatsChunkMetadata is split off from TsStatsChunk
  * because these are needed more frequently than the rest of the structure,
@@ -74,10 +82,8 @@ typedef struct TsStatsChunkMetadata
 	 * bits 2-63: seqno, incremented at every update for recency tracking
 	 */
 	pg_atomic_uint64 state;
-	/* Chunks are identified by the compressed_relid */
-	Oid compressed_relid;
-	/* The uncompressed_relid is placed here for better cache alignment */
-	Oid uncompressed_relid;
+	/* Chunks are identified by relids.compressed_relid */
+	TsStatsRelids relids;
 } TsStatsChunkMetadata;
 
 /* Bit flag values as per the above: */
@@ -328,8 +334,13 @@ mms_merge(MinMaxSumDataWA *dst, const MinMaxSumData *src)
 	uint64 cur_sqsum = pg_atomic_read_u64(&dst->sqsum_val);
 	for (int r = 0; r < TS_STATS_MMS_CAS_MAX_RETRIES; r++)
 	{
-		double new_d = *((double *) &cur_sqsum) + src->sqsum_val;
-		uint64 desired = *((uint64 *) &new_d);
+		double new_d;
+		memcpy(&new_d, &cur_sqsum, sizeof(double));
+		new_d += src->sqsum_val;
+
+		uint64 desired;
+		memcpy(&desired, &new_d, sizeof(uint64));
+
 		if (pg_atomic_compare_exchange_u64(&dst->sqsum_val, &cur_sqsum, desired))
 		{
 			break;
@@ -347,7 +358,7 @@ mms_update(MinMaxSumDataWA *dst, const MinMaxSumData *src)
 	pg_atomic_write_u32(&dst->max_val, src->max_val);
 	pg_atomic_write_u64(&dst->sum_val, src->sum_val);
 
-	/* sqsum is calculated as a double and then cast to uint64 for atomic operations */
+	/* sqsum is calculated as a double and then copied to uint64 for atomic operations */
 	pg_atomic_write_u64(&dst->sqsum_val, *((uint64 *) &src->sqsum_val));
 }
 

src/ts_stats/ts_stats_record.c

@@ -5,11 +5,13 @@
  */
 
 #include "ts_stats_record.h"
+#include "ts_stats/ts_stats_defs.h"
 #include "ts_stats_segment.h"
 
 #include <postgres.h>
 #include <miscadmin.h>
 #include <storage/lwlock.h>
+#include <utils/dsa.h>
 #include <utils/timestamp.h>
 
 static inline uint64
@@ -36,10 +38,10 @@ slot_touch_timestamps(TsStatsChunk *slot, uint64 now_us)
 }
 
 void
-ts_stats_chunk_record_compression(Oid compressed_relid, Oid uncompressed_relid,
-								  const CompressionStatsAccumulator *a)
+ts_stats_chunk_record_compression(TsStatsRelids relids, const CompressionStatsAccumulator *a)
 {
-	if (!OidIsValid(compressed_relid) || !OidIsValid(uncompressed_relid) || a->batch_count == 0)
+	if (!OidIsValid(relids.compressed_relid) || !OidIsValid(relids.uncompressed_relid) ||
+		a->batch_count == 0)
 	{
 		return;
 	}
@@ -53,8 +55,7 @@ ts_stats_chunk_record_compression(Oid compressed_relid, Oid uncompressed_relid,
 	/* get the slot_index for the upsert. this sets the in-progress flag and updates the bucket's
 	 * metadata if successful.
 	 */
-	int32 slot_idx =
-		ts_stats_chunk_segment_prepare_upsert(seg, compressed_relid, uncompressed_relid);
+	int32 slot_idx = ts_stats_chunk_segment_prepare_upsert(seg, relids);
 	if (slot_idx == TS_STATS_INVALID_IDX)
 	{
 		/* if we failed to find or insert a slot for this compressed_relid, we just give up this
@@ -85,14 +86,13 @@ ts_stats_chunk_record_compression(Oid compressed_relid, Oid uncompressed_relid,
 	slot_touch_timestamps(slot, now_us);
 
 	/* clear the in-progress flag */
-	ts_stats_chunk_segment_finish_upsert(seg, slot_idx, compressed_relid);
+	ts_stats_chunk_segment_finish_upsert(seg, slot_idx, relids.compressed_relid);
 }
 
 void
-ts_stats_chunk_record_cmd(Oid compressed_relid, Oid uncompressed_relid, CmdType cmd,
-						  const SharedCounters *c)
+ts_stats_chunk_record_cmd(TsStatsRelids relids, CmdType cmd, const SharedCounters *c)
 {
-	if (!OidIsValid(compressed_relid) || !OidIsValid(uncompressed_relid) || c == NULL)
+	if (!OidIsValid(relids.compressed_relid) || !OidIsValid(relids.uncompressed_relid) || c == NULL)
 	{
 		return;
 	}
@@ -110,8 +110,7 @@ ts_stats_chunk_record_cmd(Oid compressed_relid, Oid uncompressed_relid, CmdType
 	/* get the slot_index for the upsert. this sets the in-progress flag and updates the bucket's
 	 * metadata if successful.
 	 */
-	int32 slot_idx =
-		ts_stats_chunk_segment_prepare_upsert(seg, compressed_relid, uncompressed_relid);
+	int32 slot_idx = ts_stats_chunk_segment_prepare_upsert(seg, relids);
 	if (slot_idx == TS_STATS_INVALID_IDX)
 	{
 		/* if we failed to find or insert a slot for this compressed_relid, we just give up this
@@ -185,14 +184,13 @@ ts_stats_chunk_record_cmd(Oid compressed_relid, Oid uncompressed_relid, CmdType
 	slot_touch_timestamps(slot, now_us);
 
 	/* clear the in-progress flag */
-	ts_stats_chunk_segment_finish_upsert(seg, slot_idx, compressed_relid);
+	ts_stats_chunk_segment_finish_upsert(seg, slot_idx, relids.compressed_relid);
 }
 
 void
-ts_stats_chunk_record_decompression(Oid compressed_relid, Oid uncompressed_relid,
-									const CompressionStatsAccumulator *a)
+ts_stats_chunk_record_decompression(TsStatsRelids relids, const CompressionStatsAccumulator *a)
 {
-	if (!OidIsValid(compressed_relid) || !OidIsValid(uncompressed_relid))
+	if (!OidIsValid(relids.compressed_relid) || !OidIsValid(relids.uncompressed_relid))
 	{
 		return;
 	}
@@ -210,8 +208,7 @@ ts_stats_chunk_record_decompression(Oid compressed_relid, Oid uncompressed_relid
 	/* get the slot_index for the upsert. this sets the in-progress flag and updates the bucket's
 	 * metadata if successful.
 	 */
-	int32 slot_idx =
-		ts_stats_chunk_segment_prepare_upsert(seg, compressed_relid, uncompressed_relid);
+	int32 slot_idx = ts_stats_chunk_segment_prepare_upsert(seg, relids);
 	if (slot_idx == TS_STATS_INVALID_IDX)
 	{
 		/* if we failed to find or insert a slot for this compressed_relid, we just give up this
@@ -257,7 +254,7 @@ ts_stats_chunk_record_decompression(Oid compressed_relid, Oid uncompressed_relid
 	slot_touch_timestamps(slot, now_us);
 
 	/* clear the in-progress flag */
-	ts_stats_chunk_segment_finish_upsert(seg, slot_idx, compressed_relid);
+	ts_stats_chunk_segment_finish_upsert(seg, slot_idx, relids.compressed_relid);
 }
 
 bool
@@ -290,3 +287,52 @@ ts_stats_chunk_lookup(Oid compressed_relid, TsStatsChunk *out)
 	memcpy(out, slot, sizeof(TsStatsChunk));
 	return true;
 }
+
+void
+ts_stats_chunk_evict(Oid compressed_relid)
+{
+	if (!OidIsValid(compressed_relid))
+	{
+		return;
+	}
+	TsStatsChunkSegment *seg = ts_get_stats_chunk_segment();
+	if (seg == NULL)
+	{
+		return;
+	}
+
+	int32 slot_idx = ts_stats_chunk_segment_lookup(seg, compressed_relid);
+
+	if (slot_idx == TS_STATS_INVALID_IDX)
+	{
+		return;
+	}
+	Assert(slot_idx >= 0 && slot_idx < (int32) seg->num_slots);
+
+	TsStatsChunkMetadata *meta_base = ts_stats_chunk_metadata(seg);
+	TsStatsChunkMetadata *meta = &meta_base[slot_idx];
+
+	/* limited CAS loop to ensure we don't evict a chunk that is currently being updated, the danger
+	 * is that the one that we are evicting is already being overwritten by a newer/other chunk.
+	 */
+	uint64 state = pg_atomic_read_u64(&meta->state);
+	uint64 new_seqno = pg_atomic_add_fetch_u64(&seg->update_seqno, 1);
+	uint64 new_state = (new_seqno << TS_STATS_CHUNK_METADATA_SEQNO_SHIFT);
+	for (int i = 0;
+		 i < 5 &&
+		 /* the slot still matches the compressed_relid */
+		 meta->relids.compressed_relid == compressed_relid &&
+		 /* the slot is still valid and not in progress */
+		 TS_STATS_CHUNK_METADATA_IS_VALID(state) &&
+		 !(TS_STATS_CHUNK_METADATA_IS_IN_PROGRESS(state)) &&
+		 /* the slot being updated is older than the sequence number of the current update */
+		 TS_STATS_CHUNK_METADATA_GET_SEQNO(state) < new_seqno;
+		 ++i)
+	{
+		if (pg_atomic_compare_exchange_u64(&meta->state, &state, new_state))
+		{
+			meta->relids.compressed_relid = InvalidOid;
+			break;
+		}
+	}
+}

src/ts_stats/ts_stats_record.h

@@ -53,24 +53,26 @@ ts_stats_compression_acc_batch(CompressionStatsAccumulator *a, uint32 batch_rows
 /* Update shared memory with new statistics information. */
 
 /* Compression — commits the accumulator into the chunk's slot. */
-extern TSDLLEXPORT void ts_stats_chunk_record_compression(Oid compressed_relid,
-														  Oid uncompressed_relid,
+extern TSDLLEXPORT void ts_stats_chunk_record_compression(TsStatsRelids relids,
 														  const CompressionStatsAccumulator *a);
 
 /* Cmd stats on compressed data. cmd is one of CMD_INSERT / CMD_UPDATE / CMD_DELETE / CMD_SELECT */
-extern TSDLLEXPORT void ts_stats_chunk_record_cmd(Oid compressed_relid, Oid uncompressed_relid,
-												  CmdType cmd, const SharedCounters *c);
+extern TSDLLEXPORT void ts_stats_chunk_record_cmd(TsStatsRelids relids, CmdType cmd,
+												  const SharedCounters *c);
 
 /* For backfilling compressed data during decompression.
  * This function will update the compression stats for the chunk when it finds more complete
  * information about the compressed data than the one we have in memory. The reason we
  * need this is because the in-memory entries may miss the compression event, either because
  * of a restart or because the chunk information has been evicted from memory due to lack of use.
  */
-extern TSDLLEXPORT void ts_stats_chunk_record_decompression(Oid compressed_relid,
-															Oid uncompressed_relid,
+extern TSDLLEXPORT void ts_stats_chunk_record_decompression(TsStatsRelids relids,
 															const CompressionStatsAccumulator *a);
 
 /* Lookup TsStatsChunk based on the compressed relid. It returns true if the chunk is found, false
  * otherwise. */
 extern TSDLLEXPORT bool ts_stats_chunk_lookup(Oid compressed_relid, TsStatsChunk *out);
+
+/* Evict the chunk information for a given compressed relid. This function is called when we
+ * drop a compressed chunk. */
+extern TSDLLEXPORT void ts_stats_chunk_evict(Oid compressed_relid);