Merge branch 'master' into stable

Author: za-arthur

expected/pg_variables.out

@@ -553,6 +553,20 @@ SELECT pgv_insert('vars3', 'r1', row(1, 1));
 ERROR:  new record structure differs from variable "r1" structure
 SELECT pgv_insert('vars3', 'r1', row('str1', 'str1'));
 ERROR:  new record structure differs from variable "r1" structure
+SELECT pgv_select('vars3', 'r1') LIMIT 2;
+ pgv_select 
+------------
+ (,strNULL)
+ (1,str11)
+(2 rows)
+
+SELECT pgv_select('vars3', 'r1') LIMIT 2 OFFSET 2;
+ pgv_select 
+------------
+ (2,)
+ (0,str00)
+(2 rows)
+
 SELECT pgv_select('vars3', 'r1');
  pgv_select 
 ------------

expected/pg_variables_trans.out

@@ -1854,6 +1854,11 @@ SELECT pgv_insert('package', 'errs',row(1), true);
 
 (1 row)
 
+-- Variable should not exists in case when error occurs during creation
+SELECT pgv_insert('vars4', 'r1', row('str1', 'str1'));
+ERROR:  could not identify a hash function for type unknown
+SELECT pgv_select('vars4', 'r1', 0);
+ERROR:  unrecognized variable "r1"
 SELECT pgv_free();
  pgv_free 
 ----------

pg_variables.c

@@ -57,6 +57,7 @@ static Variable *createVariableInternal(Package *package,
 										text *name, Oid typid,
 										bool is_transactional);
 static void removePackageInternal(Package *package);
+static void resetVariablesCache(bool with_package);
 
 /* Functions to work with transactional objects */
 static void createSavepoint(TransObject *object, TransObjectType type);
@@ -67,7 +68,6 @@ static void copyValue(VarState *src, VarState *dest, Variable *destVar);
 static void freeValue(VarState *varstate, Oid typid);
 static void removeState(TransObject *object, TransObjectType type,
 						TransState *stateToDelete);
-static void removeObject(TransObject *object, TransObjectType type);
 static bool isObjectChangedInCurrentTrans(TransObject *object);
 static bool isObjectChangedInUpperTrans(TransObject *object);
 
@@ -80,6 +80,9 @@ static void makePackHTAB(Package *package, bool is_trans);
 static inline ChangedObject *makeChangedObject(TransObject *object,
 											   MemoryContext ctx);
 
+/* Hook functions */
+static void variable_ExecutorEnd(QueryDesc *queryDesc);
+
 #define CHECK_ARGS_FOR_NULL() \
 do { \
 	if (fcinfo->argnull[0]) \
@@ -99,7 +102,19 @@ static MemoryContext ModuleContext = NULL;
 static Package *LastPackage = NULL;
 /* Recent variable */
 static Variable *LastVariable = NULL;
+/* Recent row type id */
+static Oid LastTypeId = InvalidOid;
+
+/*
+ * Cache sequentially search through hash table status.  It is necessary for
+ * clean up if hash_seq_term() wasn't called or if we didn't scan the whole
+ * table. In this case we need to manually call hash_seq_term() within
+ * variable_ExecutorEnd().
+ */
+static HASH_SEQ_STATUS *LastHSeqStatus = NULL;
 
+/* Saved hook values for recall */
+static ExecutorEnd_hook_type prev_ExecutorEnd = NULL;
 
 /* This stack contains lists of changed variables and packages per each subxact level */
 static dlist_head *changesStack = NULL;
@@ -291,7 +306,7 @@ variable_insert(PG_FUNCTION_ARGS)
 
 	Oid			tupType;
 	int32		tupTypmod;
-	TupleDesc	tupdesc;
+	TupleDesc	tupdesc = NULL;
 	RecordVar  *record;
 
 	/* Checks */
@@ -360,23 +375,34 @@ variable_insert(PG_FUNCTION_ARGS)
 	/* Insert a record */
 	tupType = HeapTupleHeaderGetTypeId(rec);
 	tupTypmod = HeapTupleHeaderGetTypMod(rec);
-	tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
 
 	record = &(GetActualValue(variable).record);
 	if (!record->tupdesc)
 	{
 		/*
 		 * This is the first record for the var_name. Initialize record.
 		 */
+		tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
 		init_record(record, tupdesc, variable);
 	}
-	else
+	else if (LastTypeId == RECORDOID || !OidIsValid(LastTypeId) ||
+		LastTypeId != tupType)
+	{
+		/*
+		 * We need to check attributes of the new row if this is a transient
+		 * record type or if last record has different id.
+		 */
+		tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
 		check_attributes(variable, tupdesc);
+	}
+
+	LastTypeId = tupType;
 
 	insert_record(variable, rec);
 
 	/* Release resources */
-	ReleaseTupleDesc(tupdesc);
+	if (tupdesc)
+		ReleaseTupleDesc(tupdesc);
 
 	PG_FREE_IF_COPY(package_name, 0);
 	PG_FREE_IF_COPY(var_name, 1);
@@ -396,7 +422,6 @@ variable_update(PG_FUNCTION_ARGS)
 	bool		res;
 	Oid			tupType;
 	int32		tupTypmod;
-	TupleDesc	tupdesc;
 
 	/* Checks */
 	CHECK_ARGS_FOR_NULL();
@@ -447,14 +472,22 @@ variable_update(PG_FUNCTION_ARGS)
 	/* Update a record */
 	tupType = HeapTupleHeaderGetTypeId(rec);
 	tupTypmod = HeapTupleHeaderGetTypMod(rec);
-	tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
 
-	check_attributes(variable, tupdesc);
+	if (LastTypeId == RECORDOID || !OidIsValid(LastTypeId) ||
+		LastTypeId != tupType)
+	{
+		TupleDesc	tupdesc = NULL;
+
+		tupdesc = lookup_rowtype_tupdesc(tupType, tupTypmod);
+		check_attributes(variable, tupdesc);
+		ReleaseTupleDesc(tupdesc);
+	}
+
+	LastTypeId = tupType;
+
 	res = update_record(variable, rec);
 
 	/* Release resources */
-	ReleaseTupleDesc(tupdesc);
-
 	PG_FREE_IF_COPY(package_name, 0);
 	PG_FREE_IF_COPY(var_name, 1);
 
@@ -575,6 +608,8 @@ variable_select(PG_FUNCTION_ARGS)
 		MemoryContextSwitchTo(oldcontext);
 		PG_FREE_IF_COPY(package_name, 0);
 		PG_FREE_IF_COPY(var_name, 1);
+
+		LastHSeqStatus = rstat;
 	}
 
 	funcctx = SRF_PERCALL_SETUP();
@@ -592,6 +627,7 @@ variable_select(PG_FUNCTION_ARGS)
 	}
 	else
 	{
+		LastHSeqStatus = NULL;
 		pfree(rstat);
 		SRF_RETURN_DONE(funcctx);
 	}
@@ -867,8 +903,7 @@ remove_variable(PG_FUNCTION_ARGS)
 		GetActualState(variable)->is_valid = false;
 	}
 
-	/* Remove variable from cache */
-	LastVariable = NULL;
+	resetVariablesCache(false);
 
 	PG_FREE_IF_COPY(package_name, 0);
 	PG_FREE_IF_COPY(var_name, 1);
@@ -904,9 +939,7 @@ remove_package(PG_FUNCTION_ARGS)
 				 errmsg("unrecognized package \"%s\"", key)));
 	}
 
-	/* Remove package and variable from cache */
-	LastPackage = NULL;
-	LastVariable = NULL;
+	resetVariablesCache(true);
 
 	PG_FREE_IF_COPY(package_name, 0);
 	PG_RETURN_VOID();
@@ -934,6 +967,20 @@ removePackageInternal(Package *package)
 	GetActualState(package)->is_valid = false;
 }
 
+/*
+ * Reset cache variables to their default values. It is necessary to do in case
+ * of some changes: removing, rollbacking, etc.
+ */
+static void
+resetVariablesCache(bool with_package)
+{
+	/* Remove package and variable from cache */
+	if (with_package)
+		LastPackage = NULL;
+	LastVariable = NULL;
+	LastTypeId = InvalidOid;
+}
+
 /*
  * Remove all packages and variables.
  * Memory context will be released after committing.
@@ -955,9 +1002,7 @@ remove_packages(PG_FUNCTION_ARGS)
 		removePackageInternal(package);
 	}
 
-	/* Remove package and variable from cache */
-	LastPackage = NULL;
-	LastVariable = NULL;
+	resetVariablesCache(true);
 
 	PG_RETURN_VOID();
 }
@@ -1157,6 +1202,8 @@ get_packages_stats(PG_FUNCTION_ARGS)
 			hash_seq_init(pstat, packagesHash);
 
 			funcctx->user_fctx = pstat;
+
+			LastHSeqStatus = pstat;
 		}
 		else
 			funcctx->user_fctx = NULL;
@@ -1202,6 +1249,7 @@ get_packages_stats(PG_FUNCTION_ARGS)
 	}
 	else
 	{
+		LastHSeqStatus = NULL;
 		pfree(pstat);
 		SRF_RETURN_DONE(funcctx);
 	}
@@ -1577,13 +1625,14 @@ copyValue(VarState *src, VarState *dest, Variable *destVar)
 static void
 freeValue(VarState *varstate, Oid typid)
 {
-	if (typid == RECORDOID)
+	if (typid == RECORDOID && varstate->value.record.hctx)
 	{
 		/* All records will be freed */
 		MemoryContextDelete(varstate->value.record.hctx);
 	}
 	else if (varstate->value.scalar.typbyval == false &&
-			 varstate->value.scalar.is_null == false)
+			 varstate->value.scalar.is_null == false &&
+			 varstate->value.scalar.value)
 	{
 		pfree(DatumGetPointer(varstate->value.scalar.value));
 	}
@@ -1602,7 +1651,8 @@ removeState(TransObject *object, TransObjectType type, TransState *stateToDelete
 	pfree(stateToDelete);
 }
 
-static void
+/* Remove package or variable (either transactional or regular) */
+void
 removeObject(TransObject *object, TransObjectType type)
 {
 	bool		found;
@@ -1623,7 +1673,9 @@ removeObject(TransObject *object, TransObjectType type)
 		hash = packagesHash;
 	}
 	else
-		hash = ((Variable *) object)->package->varHashTransact;
+		hash = ((Variable *) object)->is_transactional ?
+			   ((Variable *) object)->package->varHashTransact :
+			   ((Variable *) object)->package->varHashRegular;
 
 	/* Remove all object's states */
 	while (!dlist_is_empty(&object->states))
@@ -1632,8 +1684,7 @@ removeObject(TransObject *object, TransObjectType type)
 	/* Remove object from hash table */
 	hash_search(hash, object->name, HASH_REMOVE, &found);
 
-	LastPackage = NULL;
-	LastVariable = NULL;
+	resetVariablesCache(true);
 }
 
 /*
@@ -2004,8 +2055,7 @@ processChanges(Action action)
 		MemoryContextDelete(ModuleContext);
 		packagesHash = NULL;
 		ModuleContext = NULL;
-		LastPackage = NULL;
-		LastVariable = NULL;
+		resetVariablesCache(true);
 		changesStack = NULL;
 		changesStackContext = NULL;
 	}
@@ -2065,6 +2115,23 @@ pgvTransCallback(XactEvent event, void *arg)
 	}
 }
 
+/*
+ * ExecutorEnd hook: clean up hash table sequential scan status
+ */
+static void
+variable_ExecutorEnd(QueryDesc *queryDesc)
+{
+	if (LastHSeqStatus)
+	{
+		hash_seq_term(LastHSeqStatus);
+		LastHSeqStatus = NULL;
+	}
+	if (prev_ExecutorEnd)
+		prev_ExecutorEnd(queryDesc);
+	else
+		standard_ExecutorEnd(queryDesc);
+}
+
 /*
  * Register callback function when module starts
  */
@@ -2073,6 +2140,10 @@ _PG_init(void)
 {
 	RegisterXactCallback(pgvTransCallback, NULL);
 	RegisterSubXactCallback(pgvSubTransCallback, NULL);
+
+	/* Install hooks. */
+	prev_ExecutorEnd = ExecutorEnd_hook;
+	ExecutorEnd_hook = variable_ExecutorEnd;
 }
 
 /*
@@ -2083,4 +2154,5 @@ _PG_fini(void)
 {
 	UnregisterXactCallback(pgvTransCallback, NULL);
 	UnregisterSubXactCallback(pgvSubTransCallback, NULL);
+	ExecutorEnd_hook = prev_ExecutorEnd;
 }

pg_variables.h

@@ -155,6 +155,7 @@ extern void check_record_key(Variable *variable, Oid typid);
 extern void insert_record(Variable *variable, HeapTupleHeader tupleHeader);
 extern bool update_record(Variable *variable, HeapTupleHeader tupleHeader);
 extern bool delete_record(Variable *variable, Datum value, bool is_null);
+extern void removeObject(TransObject *object, TransObjectType type);
 
 #define GetActualState(object) \
 	(dlist_head_element(TransState, node, &((TransObject *) object)->states))

pg_variables_record.c

@@ -81,17 +81,29 @@ init_record(RecordVar *record, TupleDesc tupdesc, Variable *variable)
 								 TYPECACHE_HASH_PROC_FINFO |
 								 TYPECACHE_CMP_PROC_FINFO);
 
+	/*
+	 * In case something went wrong, you need to roll back the changes before
+	 * completing the transaction, because the variable may be regular
+	 * and not present in list of changed vars.
+	 */
 	if (!OidIsValid(typentry->hash_proc_finfo.fn_oid))
+	{
+		/* At this point variable is just created, so we simply remove it. */
+		removeObject(&variable->transObject, TRANS_VARIABLE);
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_FUNCTION),
 				 errmsg("could not identify a hash function for type %s",
 						format_type_be(keyid))));
+	}
 
 	if (!OidIsValid(typentry->cmp_proc_finfo.fn_oid))
+	{
+		removeObject(&variable->transObject, TRANS_VARIABLE);
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_FUNCTION),
 				 errmsg("could not identify a matching function for type %s",
 						format_type_be(keyid))));
+	}
 
 	/* Initialize the record */
 

sql/pg_variables.sql

@@ -153,6 +153,9 @@ SELECT pgv_insert('vars3', 'r1', row(1, 'str1', 'str2'));
 SELECT pgv_insert('vars3', 'r1', row(1, 1));
 SELECT pgv_insert('vars3', 'r1', row('str1', 'str1'));
 
+SELECT pgv_select('vars3', 'r1') LIMIT 2;
+SELECT pgv_select('vars3', 'r1') LIMIT 2 OFFSET 2;
+
 SELECT pgv_select('vars3', 'r1');
 SELECT pgv_select('vars3', 'r1', 1);
 SELECT pgv_select('vars3', 'r1', 0);