fss_neighbours saves fs

Author: Alexandra Pervushina

aqo_shared.c

@@ -200,12 +200,14 @@ aqo_init_shmem(void)
 		aqo_state->stat_changed = false;
 		aqo_state->data_changed = false;
 		aqo_state->queries_changed = false;
+		aqo_state->neighbours_changed = false;
 
 		LWLockInitialize(&aqo_state->lock, LWLockNewTrancheId());
 		LWLockInitialize(&aqo_state->stat_lock, LWLockNewTrancheId());
 		LWLockInitialize(&aqo_state->qtexts_lock, LWLockNewTrancheId());
 		LWLockInitialize(&aqo_state->data_lock, LWLockNewTrancheId());
 		LWLockInitialize(&aqo_state->queries_lock, LWLockNewTrancheId());
+		LWLockInitialize(&aqo_state->neighbours_lock, LWLockNewTrancheId());
 	}
 
 	/* Init shared memory hash for partial aqo data table*/
@@ -241,7 +243,7 @@ aqo_init_shmem(void)
 
 	/* Shared memory hash table for fss neighbours */
 	info.keysize = sizeof(int);
-	info.entrysize = sizeof(DataEntry*);
+	info.entrysize = sizeof(uint64);
 	fss_neighbours = ShmemInitHash("AQO fss neighbours HTAB", fss_max_items, fss_max_items,
 							  &info, HASH_ELEM | HASH_BLOBS);
 
@@ -252,6 +254,7 @@ aqo_init_shmem(void)
 	LWLockRegisterTranche(aqo_state->qtext_trancheid, "AQO Query Texts Tranche");
 	LWLockRegisterTranche(aqo_state->data_lock.tranche, "AQO Data Lock Tranche");
 	LWLockRegisterTranche(aqo_state->queries_lock.tranche, "AQO Queries Lock Tranche");
+	LWLockRegisterTranche(aqo_state->neighbours_lock.tranche, "AQO Neighbours Lock Tranche");
 
 	if (!IsUnderPostmaster && !found)
 	{
@@ -293,7 +296,7 @@ aqo_memsize(void)
 	size = add_size(size, hash_estimate_size(fs_max_items, sizeof(QueryTextEntry)));
 	size = add_size(size, hash_estimate_size(fss_max_items, sizeof(DataEntry)));
 	size = add_size(size, hash_estimate_size(fs_max_items, sizeof(QueriesEntry)));
-	size = add_size(size, hash_estimate_size(fss_max_items, sizeof(DataEntry*)));
+	size = add_size(size, hash_estimate_size(fss_max_items, sizeof(uint64)));
 
 	return size;
 }

aqo_shared.h

@@ -42,6 +42,9 @@ typedef struct AQOSharedState
 
 	LWLock		queries_lock;  /* lock for access to queries storage */
 	bool		queries_changed;
+
+	LWLock		neighbours_lock;  /* lock for access to queries storage */
+	bool		neighbours_changed;
 } AQOSharedState;
 
 

storage.c

@@ -1302,7 +1302,7 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 	bool		tblOverflow;
 	HASHACTION	action;
 	bool		result;
-	DataEntry**  prev;
+	uint64     *prev_fs;
 
 	Assert(!LWLockHeldByMe(&aqo_state->data_lock));
 
@@ -1387,29 +1387,41 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 	Assert(ptr != NULL);
 
 	/*
-	*  Find prev fss
+	*  Find prev fs with the same fss
 	*/
 
-	prev = (DataEntry **) hash_search(fss_neighbours, &fss, HASH_ENTER, &found);
+	LWLockAcquire(&aqo_state->neighbours_lock, LW_EXCLUSIVE);
+
+	prev_fs = (uint64 *) hash_search(fss_neighbours, &fss, HASH_ENTER, &found);
 	if (!found)
 	{
-		prev = memcpy(palloc(sizeof(DataEntry **)), entry, sizeof(DataEntry **));
-		//*prev = entry;
+		//*prev_fs = fs;
+		memcpy(prev_fs, &entry->key.fs, sizeof(uint64));
 	}
 	else
 	{
-		(*prev)->list.next_fs = fss;
-		neighbour_list.prev_fs = (*prev)->key.fs;
+		data_key	prev_key = {.fs = *prev_fs, .fss = fss};
+		DataEntry  *prev;
+
+		prev = (DataEntry *) hash_search(data_htab, &prev_key, HASH_FIND, NULL);
+		//prev->list.next_fs = fs;
+		memcpy(&prev->list.next_fs, &fs, sizeof(uint64));
+		neighbour_list.prev_fs = prev->key.fs;
 	}
 
+	prev_fs = (uint64 *) hash_search(fss_neighbours, &fss, HASH_FIND, &found);
+	elog(NOTICE, "%d 🗿🔫", found);
+
+	LWLockRelease(&aqo_state->neighbours_lock);
+
 
 	/*
 	 * Copy AQO data into allocated DSA segment
 	 */
 
 	memcpy(ptr, &key, sizeof(data_key)); /* Just for debug */
 	ptr += sizeof(data_key);
-	memcpy(ptr,&neighbour_list, sizeof(fs_list));
+	memcpy(ptr, &neighbour_list, sizeof(fs_list));
 	ptr += sizeof(fs_list);
 	if (entry->cols > 0)
 	{

storage.h

@@ -57,8 +57,8 @@ typedef struct data_key
 
 typedef struct fs_list
 {
-	int64	prev_fs;
-	int64	next_fs;
+	uint64	prev_fs;
+	uint64	next_fs;
 } fs_list;
 
 typedef struct DataEntry