Fixed processing of the rewrite.max_xlock_time parameter.

1. So far, the value of the parameter was not passed to the worker at all.

2. The worker coding was such that the parameter was only checked during
   logical decoding, but not when applying the data changes.

Author: Antonin Houska

concurrent.c

@@ -27,7 +27,8 @@ static void apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 									 Relation ident_index,
 									 TupleTableSlot	*slot_dst_ind,
 									 partitions_hash *partitions,
-									 TupleConversionMapExt *conv_map);
+									 TupleConversionMapExt *conv_map,
+									 struct timeval *must_complete);
 static void apply_insert(HeapTuple tup, TupleTableSlot *slot,
 						 EState *estate, ModifyTableState *mtstate,
 						 struct PartitionTupleRouting *proute,
@@ -106,6 +107,20 @@ pg_rewrite_process_concurrent_changes(EState *estate,
 			partitions && proute));
 
 	dstate = (DecodingOutputState *) ctx->output_writer_private;
+
+	/*
+	 * If some changes could not be applied due to time constraint, make sure
+	 * the tuplestore is empty before we insert new tuples into it.
+	 */
+	if (dstate->nchanges > 0)
+		apply_concurrent_changes(estate, mtstate, proute,
+								 dstate, ident_key, ident_key_nentries,
+								 ident_index, slot_dst_ind,
+								 partitions, conv_map, must_complete);
+	/* Ran out of time? */
+	if (dstate->nchanges > 0)
+		return false;
+
 	done = false;
 	while (!done)
 	{
@@ -129,7 +144,10 @@ pg_rewrite_process_concurrent_changes(EState *estate,
 		apply_concurrent_changes(estate, mtstate, proute,
 								 dstate, ident_key, ident_key_nentries,
 								 ident_index, slot_dst_ind,
-								 partitions, conv_map);
+								 partitions, conv_map, must_complete);
+		/* Ran out of time? */
+		if (dstate->nchanges > 0)
+			return false;
 	}
 
 	return true;
@@ -233,7 +251,8 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 						 Relation ident_index,
 						 TupleTableSlot	*slot_dst_ind,
 						 partitions_hash *partitions,
-						 TupleConversionMapExt *conv_map)
+						 TupleConversionMapExt *conv_map,
+						 struct timeval *must_complete)
 {
 	BulkInsertState	bistate = NULL;
 	HeapTuple	tup_old = NULL;
@@ -272,6 +291,9 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 		bool		isnull[1];
 		Datum		values[1];
 
+		Assert(dstate->nchanges > 0);
+		dstate->nchanges--;
+
 		/* Get the change from the single-column tuple. */
 		tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree);
 		heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull);
@@ -320,10 +342,22 @@ apply_concurrent_changes(EState *estate, ModifyTableState *mtstate,
 		/* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */
 		Assert(shouldFree);
 		pfree(tup_change);
+
+		/*
+		 * If there is a limit on the time of completion, check it
+		 * now. However, make sure the loop does not break if tup_old was set
+		 * in the previous iteration. In such a case we could not resume the
+		 * processing in the next call.
+		 */
+		if (must_complete && tup_old == NULL &&
+			processing_time_elapsed(must_complete))
+			/* The next call will process the remaining changes. */
+			break;
 	}
 
-	tuplestore_clear(dstate->tstore);
-	dstate->nchanges = 0;
+	/* If we could not apply all the changes, the next call will do. */
+	if (dstate->nchanges == 0)
+		tuplestore_clear(dstate->tstore);
 
 	PopActiveSnapshot();
 

pg_rewrite.c

@@ -453,6 +453,7 @@ get_task(int *idx, char *relschema, char *relname, bool nowait)
 	task->elevel = -1;
 
 	task->nowait = nowait;
+	task->max_xlock_time = rewrite_max_xlock_time;
 
 	*idx = i;
 	return task;
@@ -2395,12 +2396,13 @@ perform_final_merge(EState *estate,
 	{
 		int64		usec;
 		struct timeval t_start;
+		int		max_xlock_time = MyWorkerTask->max_xlock_time;
 
 		gettimeofday(&t_start, NULL);
 		/* Add the whole seconds. */
-		t_end.tv_sec = t_start.tv_sec + rewrite_max_xlock_time / 1000;
+		t_end.tv_sec = t_start.tv_sec + max_xlock_time / 1000;
 		/* Add the rest, expressed in microseconds. */
-		usec = t_start.tv_usec + 1000 * (rewrite_max_xlock_time % 1000);
+		usec = t_start.tv_usec + 1000 * (max_xlock_time % 1000);
 		/* The number of microseconds could have overflown. */
 		t_end.tv_sec += usec / USECS_PER_SEC;
 		t_end.tv_usec = usec % USECS_PER_SEC;
@@ -2497,6 +2499,11 @@ perform_final_merge(EState *estate,
 											  partitions,
 											  conv_map,
 											  NULL);
+
+		/* No time constraint, all changes must have been processed. */
+		Assert(((DecodingOutputState *)
+				ctx->output_writer_private)->nchanges == 0);
+
 	}
 
 	list_free(indexes);

pg_rewrite.h

@@ -171,6 +171,8 @@ typedef struct WorkerTask
 	 * backend might be interested in 'msg' and 'msg_detail'.
 	 */
 	bool	nowait;
+
+	int             max_xlock_time;
 } WorkerTask;
 
 #define		MAX_TASKS	8