Add fss_neighbours hash table

Author: Alexandra Pervushina

aqo--1.4--1.5.sql

@@ -21,7 +21,6 @@ DROP TABLE public.aqo_query_texts CASCADE;
 DROP TABLE public.aqo_query_stat CASCADE;
 DROP FUNCTION invalidate_deactivated_queries_cache;
 
-
 /*
  * VIEWs to discover AQO data.
  */
@@ -63,7 +62,9 @@ CREATE FUNCTION aqo_data (
   OUT features		double precision[][],
   OUT targets		double precision[],
   OUT reliability	double precision[],
-  OUT oids			Oid[]
+  OUT oids			Oid[],
+  OUT prev_fs bigint,
+  OUT next_fs bigint
 )
 RETURNS SETOF record
 AS 'MODULE_PATHNAME', 'aqo_data'

aqo_shared.c

@@ -182,6 +182,7 @@ aqo_init_shmem(void)
 	qtexts_htab = NULL;
 	data_htab = NULL;
 	queries_htab = NULL;
+	fss_neighbours = NULL;
 
 	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
 	aqo_state = ShmemInitStruct("AQO", sizeof(AQOSharedState), &found);
@@ -207,6 +208,7 @@ aqo_init_shmem(void)
 		LWLockInitialize(&aqo_state->queries_lock, LWLockNewTrancheId());
 	}
 
+	/* Init shared memory hash for partial aqo data table*/
 	info.keysize = sizeof(htab_key);
 	info.entrysize = sizeof(htab_entry);
 	fss_htab = ShmemInitHash("AQO hash",
@@ -237,6 +239,12 @@ aqo_init_shmem(void)
 	queries_htab = ShmemInitHash("AQO Queries HTAB", fs_max_items, fs_max_items,
 								 &info, HASH_ELEM | HASH_BLOBS);
 
+	/* Shared memory hash table for fss neighbours */
+	info.keysize = sizeof(int);
+	info.entrysize = sizeof(DataEntry*);
+	fss_neighbours = ShmemInitHash("AQO fss neighbours HTAB", fss_max_items, fss_max_items,
+							  &info, HASH_ELEM | HASH_BLOBS);
+
 	LWLockRelease(AddinShmemInitLock);
 	LWLockRegisterTranche(aqo_state->lock.tranche, "AQO");
 	LWLockRegisterTranche(aqo_state->stat_lock.tranche, "AQO Stat Lock Tranche");
@@ -285,6 +293,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*)));
 
 	return size;
 }

storage.c

@@ -52,7 +52,7 @@ typedef enum {
 
 typedef enum {
 	AD_FS = 0, AD_FSS, AD_NFEATURES, AD_FEATURES, AD_TARGETS, AD_RELIABILITY,
-	AD_OIDS, AD_TOTAL_NCOLS
+	AD_OIDS, AD_PREV_FS, AD_NEXT_FS, AD_TOTAL_NCOLS
 } aqo_data_cols;
 
 typedef enum {
@@ -74,6 +74,7 @@ dsa_area *qtext_dsa = NULL;
 HTAB *data_htab = NULL;
 dsa_area *data_dsa = NULL;
 HTAB *deactivated_queries = NULL;
+HTAB *fss_neighbours = NULL;
 
 /* Used to check data file consistency */
 static const uint32 PGAQO_FILE_HEADER = 123467589;
@@ -1293,13 +1294,15 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 	DataEntry  *entry;
 	bool		found;
 	data_key	key = {.fs = fs, .fss = fss};
+	fs_list     neighbour_list = {0};
 	int			i;
 	char	   *ptr;
 	ListCell   *lc;
 	size_t		size;
 	bool		tblOverflow;
 	HASHACTION	action;
 	bool		result;
+	DataEntry**  prev;
 
 	Assert(!LWLockHeldByMe(&aqo_state->data_lock));
 
@@ -1383,12 +1386,32 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 	ptr = (char *) dsa_get_address(data_dsa, entry->data_dp);
 	Assert(ptr != NULL);
 
+	/*
+	*  Find prev fss
+	*/
+
+	prev = (DataEntry **) hash_search(fss_neighbours, &fss, HASH_ENTER, &found);
+	if (!found)
+	{
+		*prev = entry;
+	}
+	else
+	{
+		(*prev)->list.next_fs = fss;
+	}
+	neighbour_list.prev_fs = (*prev)->key.fs;
+
+	prev = (DataEntry **) hash_search(fss_neighbours, &fss, HASH_FIND, &found);
+	elog(NOTICE, "%d", found);
+
 	/*
 	 * 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));
+	ptr += sizeof(fs_list);
 	if (entry->cols > 0)
 	{
 		for (i = 0; i < entry->rows; i++)
@@ -1652,6 +1675,8 @@ aqo_data(PG_FUNCTION_ARGS)
 		values[AD_FS] = Int64GetDatum(entry->key.fs);
 		values[AD_FSS] = Int32GetDatum((int) entry->key.fss);
 		values[AD_NFEATURES] = Int32GetDatum(entry->cols);
+		values[AD_PREV_FS] = Int64GetDatum(entry->list.prev_fs);
+		values[AD_NEXT_FS] = Int64GetDatum(entry->list.next_fs);
 
 		/* Fill values from the DSA data chunk */
 		Assert(DsaPointerIsValid(entry->data_dp));
@@ -2135,6 +2160,8 @@ cleanup_aqo_database(bool gentle, int *fs_num, int *fss_num)
 		foreach(lc, junk_fss)
 		{
 			data_key	key = {.fs = entry->fs, .fss = lfirst_int(lc)};
+			//TODO fix fs list
+			hash_search(fss_neighbours, &lfirst_int(lc), HASH_REMOVE, NULL);
 			(*fss_num) += (int) _aqo_data_remove(&key);
 		}
 

storage.h

@@ -55,9 +55,16 @@ typedef struct data_key
 	int64	fss; /* just for alignment */
 } data_key;
 
+typedef struct fs_list
+{
+	int64	prev_fs;
+	int64	next_fs;
+} fs_list;
+
 typedef struct DataEntry
 {
 	data_key key;
+	fs_list list;
 
 	/* defines a size and data placement in the DSA memory block */
 	int cols; /* aka nfeatures */
@@ -89,6 +96,7 @@ extern HTAB *stat_htab;
 extern HTAB *qtexts_htab;
 extern HTAB *queries_htab; /* TODO */
 extern HTAB *data_htab; /* TODO */
+extern HTAB *fss_neighbours;
 
 extern StatEntry *aqo_stat_store(uint64 queryid, bool use_aqo, double plan_time,
 								 double exec_time, double est_error);