Be more careful with locks of relations and syscaches in get_list_of_relids() routine

Switch on feature 'search on neigur feature spaces' by a GUC (disabled by default).

Author: Alexandra Pervushina

aqo.c

@@ -213,6 +213,19 @@ _PG_init(void)
 NULL
 );
 
+DefineCustomBoolVariable(
+"aqo.wide_search",
+"Search ML data in neigur feature spaces.",
+NULL,
+&use_wide_search,
+false,
+PGC_USERSET,
+0,
+NULL,
+NULL,
+NULL
+);
+
 DefineCustomIntVariable("aqo.join_threshold",
 "Sets the threshold of number of JOINs in query beyond which AQO is used.",
 NULL,

aqo.h

@@ -173,6 +173,7 @@ extern bool	force_collect_stat;
 extern bool aqo_show_hash;
 extern bool aqo_show_details;
 extern int aqo_join_threshold;
+extern bool use_wide_search;
 
 /* Parameters for current query */
 typedef struct QueryContextData

aqo_shared.c

@@ -191,16 +191,18 @@ aqo_init_shmem(void)
 {
 /* First time through ... */
 
-LWLockInitialize(&aqo_state->lock, LWLockNewTrancheId());
 aqo_state->dsm_handler = DSM_HANDLE_INVALID;
-
 aqo_state->qtexts_dsa_handler = DSM_HANDLE_INVALID;
+aqo_state->data_dsa_handler = DSM_HANDLE_INVALID;
+
 aqo_state->qtext_trancheid = LWLockNewTrancheId();
+
 aqo_state->qtexts_changed = false;
-aqo_state->data_dsa_handler = DSM_HANDLE_INVALID;
+aqo_state->stat_changed = false;
 aqo_state->data_changed = false;
 aqo_state->queries_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());
@@ -245,7 +247,7 @@ aqo_init_shmem(void)
 LWLockRegisterTranche(aqo_state->data_lock.tranche, "AQO Data Lock Tranche");
 LWLockRegisterTranche(aqo_state->queries_lock.tranche, "AQO Queries Lock Tranche");
 
-if (!IsUnderPostmaster)
+if (!IsUnderPostmaster && !found)
 {
 before_shmem_exit(on_shmem_shutdown, (Datum) 0);
 
@@ -261,8 +263,16 @@ aqo_init_shmem(void)
 static void
 on_shmem_shutdown(int code, Datum arg)
 {
+Assert(!IsUnderPostmaster);
+
+/*
+* Save ML data to a permanent storage. Do it on postmaster shutdown only
+* to save time. We can't do so for query_texts and aqo_data because of DSM
+* limits.
+*/
 aqo_stat_flush();
 aqo_queries_flush();
+return;
 }
 
 Size

cardinality_estimation.c

@@ -24,6 +24,9 @@
 #include "machine_learning.h"
 #include "storage.h"
 
+
+bool use_wide_search = false;
+
 #ifdef AQO_DEBUG_PRINT
 static void
 predict_debug_output(List *clauses, List *selectivities,
@@ -90,7 +93,7 @@ predict_for_relation(List *clauses, List *selectivities, List *relsigns,
 */
 
 /* Try to search in surrounding feature spaces for the same node */
-if (!load_aqo_data(query_context.fspace_hash, *fss, data, NULL, true))
+if (!load_aqo_data(query_context.fspace_hash, *fss, data, NULL, use_wide_search))
 result = -1;
 else
 {

conf.add

@@ -1,3 +1,4 @@
 autovacuum = off
 shared_preload_libraries = 'postgres_fdw, aqo'
 max_parallel_workers_per_gather = 1 # switch off parallel workers because of unsteadiness
+aqo.wide_search = 'on'

path_utils.c

@@ -155,6 +155,8 @@ hashTempTupleDesc(TupleDesc desc)
 return s;
 }
 
+#include "storage/lmgr.h"
+
 /*
 * Get list of relation indexes and prepare list of permanent table reloids,
 * list of temporary table reloids (can be changed between query launches) and
@@ -177,6 +179,8 @@ get_list_of_relids(PlannerInfo *root, Relids relids, RelSortOut *rels)
 HeapTuple htup;
 Form_pg_class	classForm;
 char *relname = NULL;
+Oid relrewrite;
+char relpersistence;
 
 entry = planner_rt_fetch(index, root);
 
@@ -191,34 +195,39 @@ get_list_of_relids(PlannerInfo *root, Relids relids, RelSortOut *rels)
 if (!HeapTupleIsValid(htup))
 elog(PANIC, "cache lookup failed for reloid %u", entry->relid);
 
+/* Copy the fields from syscache and release the slot as quickly as possible. */
 classForm = (Form_pg_class) GETSTRUCT(htup);
+relpersistence = classForm->relpersistence;
+relrewrite = classForm->relrewrite;
+relname = pstrdup(NameStr(classForm->relname));
+ReleaseSysCache(htup);
 
-if (classForm->relpersistence == RELPERSISTENCE_TEMP)
+if (relpersistence == RELPERSISTENCE_TEMP)
 {
 /* The case of temporary table */
 
-Relation trel = relation_open(entry->relid, NoLock);
-TupleDesc tdesc = RelationGetDescr(trel);
+Relation	trel;
+TupleDesc	tdesc;
 
+trel = relation_open(entry->relid, NoLock);
+tdesc = RelationGetDescr(trel);
+Assert(CheckRelationLockedByMe(trel, AccessShareLock, true));
 hashes = lappend_uint64(hashes, hashTempTupleDesc(tdesc));
 relation_close(trel, NoLock);
 }
 else
 {
 /* The case of regular table */
 relname = quote_qualified_identifier(
-get_namespace_name(get_rel_namespace(entry->relid)),
-classForm->relrewrite ?
-get_rel_name(classForm->relrewrite) :
-NameStr(classForm->relname));
+get_namespace_name(get_rel_namespace(entry->relid)),
+relrewrite ? get_rel_name(relrewrite) : relname);
+
 hashes = lappend_uint64(hashes, DatumGetInt64(hash_any_extended(
 (unsigned char *) relname,
 strlen(relname), 0)));
 
 hrels = lappend_oid(hrels, entry->relid);
 }
-
-ReleaseSysCache(htup);
 }
 
 rels->hrels = list_concat(rels->hrels, hrels);

postprocessing.c

@@ -171,7 +171,6 @@ restore_selectivities(List *clauselist, List *relidslist, JoinType join_type,
 {
 List *lst = NIL;
 ListCell	*l;
-int i = 0;
 bool parametrized_sel;
 int nargs;
 int *args_hash;
@@ -220,7 +219,6 @@ restore_selectivities(List *clauselist, List *relidslist, JoinType join_type,
 Assert(cur_sel > 0);
 
 lst = lappend(lst, cur_sel);
-i++;
 }
 
 if (parametrized_sel)

storage.c

@@ -302,7 +302,9 @@ aqo_stat_store(uint64 queryid, bool use_aqo,
 entry->exec_time[pos] = exec_time;
 entry->est_error[pos] = est_error;
 }
+
 entry = memcpy(palloc(sizeof(StatEntry)), entry, sizeof(StatEntry));
+aqo_state->stat_changed = true;
 LWLockRelease(&aqo_state->stat_lock);
 return entry;
 }
@@ -424,14 +426,24 @@ aqo_stat_flush(void)
 int ret;
 long entries;
 
-LWLockAcquire(&aqo_state->stat_lock, LW_SHARED);
+/* Use exclusive lock to prevent concurrent flushing in different backends. */
+LWLockAcquire(&aqo_state->stat_lock, LW_EXCLUSIVE);
+
+if (!aqo_state->stat_changed)
+/* Hash table wasn't changed, meaningless to store it in permanent storage */
+goto end;
+
 entries = hash_get_num_entries(stat_htab);
 hash_seq_init(&hash_seq, stat_htab);
 ret = data_store(PGAQO_STAT_FILE, _form_stat_record_cb, entries,
 (void *) &hash_seq);
 if (ret != 0)
 hash_seq_term(&hash_seq);
+else
+/* Hash table and disk storage are now consistent */
+aqo_state->stat_changed = false;
 
+end:
 LWLockRelease(&aqo_state->stat_lock);
 }
 
@@ -468,7 +480,7 @@ aqo_qtexts_flush(void)
 long entries;
 
 dsa_init();
-LWLockAcquire(&aqo_state->qtexts_lock, LW_SHARED);
+LWLockAcquire(&aqo_state->qtexts_lock, LW_EXCLUSIVE);
 
 if (!aqo_state->qtexts_changed)
 /* XXX: mull over forced mode. */
@@ -480,7 +492,9 @@ aqo_qtexts_flush(void)
 (void *) &hash_seq);
 if (ret != 0)
 hash_seq_term(&hash_seq);
-aqo_state->qtexts_changed = false;
+else
+/* Hash table and disk storage are now consistent */
+aqo_state->qtexts_changed = false;
 
 end:
 LWLockRelease(&aqo_state->qtexts_lock);
@@ -530,7 +544,7 @@ aqo_data_flush(void)
 long entries;
 
 dsa_init();
-LWLockAcquire(&aqo_state->data_lock, LW_SHARED);
+LWLockAcquire(&aqo_state->data_lock, LW_EXCLUSIVE);
 
 if (!aqo_state->data_changed)
 /* XXX: mull over forced mode. */
@@ -547,6 +561,7 @@ aqo_data_flush(void)
 */
 hash_seq_term(&hash_seq);
 else
+/* Hash table and disk storage are now consistent */
 aqo_state->data_changed = false;
 end:
 LWLockRelease(&aqo_state->data_lock);
@@ -573,14 +588,22 @@ aqo_queries_flush(void)
 int ret;
 long entries;
 
-LWLockAcquire(&aqo_state->queries_lock, LW_SHARED);
+LWLockAcquire(&aqo_state->queries_lock, LW_EXCLUSIVE);
+
+if (!aqo_state->queries_changed)
+goto end;
+
 entries = hash_get_num_entries(queries_htab);
 hash_seq_init(&hash_seq, queries_htab);
 ret = data_store(PGAQO_QUERIES_FILE, _form_queries_record_cb, entries,
 (void *) &hash_seq);
 if (ret != 0)
 hash_seq_term(&hash_seq);
+else
+/* Hash table and disk storage are now consistent */
+aqo_state->queries_changed = false;
 
+end:
 LWLockRelease(&aqo_state->queries_lock);
 }
 
@@ -620,7 +643,8 @@ data_store(const char *filename, form_record_t callback,
 goto error;
 }
 
-(void) durable_rename(tmpfile, filename, LOG);
+/* Parallel (re)writing into a file haven't happen. */
+(void) durable_rename(tmpfile, filename, PANIC);
 elog(LOG, "[AQO] %d records stored in file %s.", counter, filename);
 return 0;
 
@@ -838,7 +862,7 @@ aqo_queries_load(void)
 
 LWLockAcquire(&aqo_state->queries_lock, LW_EXCLUSIVE);
 
-/* Load on postmaster sturtup. So no any concurrent actions possible here. */
+/* Load on postmaster startup. So no any concurrent actions possible here. */
 Assert(hash_get_num_entries(queries_htab) == 0);
 
 data_load(PGAQO_QUERIES_FILE, _deform_queries_record_cb, NULL);
@@ -925,6 +949,9 @@ data_load(const char *filename, deform_record_t callback, void *ctx)
 static void
 on_shmem_shutdown(int code, Datum arg)
 {
+/*
+* XXX: It can be expensive to rewrite a file on each shutdown of a backend.
+*/
 aqo_qtexts_flush();
 aqo_data_flush();
 }
@@ -1200,6 +1227,7 @@ _aqo_data_remove(data_key *key)
 
 if (hash_search(data_htab, key, HASH_REMOVE, NULL) == NULL)
 elog(PANIC, "[AQO] Inconsistent data hash table");
+
 aqo_state->data_changed = true;
 }
 
@@ -1269,8 +1297,9 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 char *ptr;
 ListCell *lc;
 size_t size;
-bool tblOverflow;
-HASHACTION action;
+bool tblOverflow;
+HASHACTION	action;
+bool result;
 
 Assert(!LWLockHeldByMe(&aqo_state->data_lock));
 
@@ -1321,7 +1350,6 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 }
 
 Assert(DsaPointerIsValid(entry->data_dp));
-Assert(entry->rows <= data->rows); /* Reserved for the future features */
 
 if (entry->cols != data->cols || entry->nrels != list_length(reloids))
 {
@@ -1387,8 +1415,9 @@ aqo_data_store(uint64 fs, int fss, OkNNrdata *data, List *reloids)
 
 aqo_state->data_changed = true;
 end:
+result = aqo_state->data_changed;
 LWLockRelease(&aqo_state->data_lock);
-return aqo_state->data_changed;
+return result;
 }
 
 static void
@@ -1496,7 +1525,7 @@ load_aqo_data(uint64 fs, int fss, OkNNrdata *data, List **reloids,
 
 dsa_init();
 
-LWLockAcquire(&aqo_state->data_lock, LW_EXCLUSIVE);
+LWLockAcquire(&aqo_state->data_lock, LW_SHARED);
 
 if (!wideSearch)
 {
@@ -1631,7 +1660,8 @@ aqo_data(PG_FUNCTION_ARGS)
 ptr += sizeof(data_key);
 
 if (entry->cols > 0)
-values[AD_FEATURES] = PointerGetDatum(form_matrix((double *)ptr, entry->rows, entry->cols));
+values[AD_FEATURES] = PointerGetDatum(form_matrix((double *) ptr,
+entry->rows, entry->cols));
 else
 nulls[AD_FEATURES] = true;
 
@@ -1719,7 +1749,9 @@ aqo_data_reset(void)
 elog(ERROR, "[AQO] hash table corrupted");
 num_remove++;
 }
-aqo_state->data_changed = true;
+
+if (num_remove > 0)
+aqo_state->data_changed = true;
 LWLockRelease(&aqo_state->data_lock);
 if (num_remove != num_entries)
 elog(ERROR, "[AQO] Query ML memory storage is corrupted or parallel access without a lock has detected.");
@@ -1831,6 +1863,7 @@ aqo_queries_store(uint64 queryid,
 entry->use_aqo = use_aqo;
 entry->auto_tuning = auto_tuning;
 
+aqo_state->queries_changed = true;
 LWLockRelease(&aqo_state->queries_lock);
 return true;
 }
@@ -1856,7 +1889,10 @@ aqo_queries_reset(void)
 elog(ERROR, "[AQO] hash table corrupted");
 num_remove++;
 }
-aqo_state->queries_changed = true;
+
+if (num_remove > 0)
+aqo_state->queries_changed = true;
+
 LWLockRelease(&aqo_state->queries_lock);
 
 if (num_remove != num_entries - 1)