pgbench-aggregated.diff

diff --git a/contrib/pgbench/pgbench.c b/contrib/pgbench/pgbench.c
index 00cab73..9fede3c 100644
--- a/contrib/pgbench/pgbench.c
+++ b/contrib/pgbench/pgbench.c
@@ -145,6 +145,8 @@ char	   *index_tablespace = NULL;
 #define naccounts	100000
 
 bool		use_log;			/* log transaction latencies to a file */
+bool		use_log_agg;		/* log aggregates instead of individual transactions */
+int			naggseconds = 1;	/* number of seconds for aggregation */
 bool		is_connect;			/* establish connection for each transaction */
 bool		is_latencies;		/* report per-command latencies */
 int			main_pid;			/* main process id used in log filename */
@@ -240,6 +242,18 @@ typedef struct
 	char	   *argv[MAX_ARGS]; /* command word list */
 } Command;
 
+typedef struct
+{
+
+	long	start_time;			/* when does the interval start */
+	int 	cnt;				/* number of transactions */
+	double	min_duration;		/* min/max durations */
+	double	max_duration;
+	double	sum;				/* sum(duration), sum(duration^2) - for estimates */
+	double	sum2;
+	
+} AggVals;
+
 static Command **sql_files[MAX_FILES];	/* SQL script files */
 static int	num_files;			/* number of script files */
 static int	num_commands = 0;	/* total number of Command structs */
@@ -364,6 +378,7 @@ usage(void)
 		   "  -f FILENAME  read transaction script from FILENAME\n"
 		   "  -j NUM       number of threads (default: 1)\n"
 		   "  -l           write transaction times to log file\n"
+		   "  -A NUM       write aggregates (over NUM seconds) instead\n"
 		   "  -M simple|extended|prepared\n"
 		   "               protocol for submitting queries to server (default: simple)\n"
 		   "  -n           do not run VACUUM before tests\n"
@@ -817,9 +832,22 @@ clientDone(CState *st, bool ok)
 	return false;				/* always false */
 }
 
+static
+void agg_vals_init(AggVals * aggs, instr_time start)
+{
+	aggs->cnt = 0;
+	aggs->sum = 0;
+	aggs->sum2 = 0;
+	
+	aggs->min_duration = 3600 * 1000000.0; /* one hour */
+	aggs->max_duration = 0;
+
+	aggs->start_time = INSTR_TIME_GET_DOUBLE(start);
+}
+
 /* return false iff client should be disconnected */
 static bool
-doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile)
+doCustom(TState *thread, CState *st, instr_time *conn_time, FILE *logfile, AggVals * agg)
 {
 	PGresult   *res;
 	Command   **commands;
@@ -881,17 +909,65 @@ top:
 			diff = now;
 			INSTR_TIME_SUBTRACT(diff, st->txn_begin);
 			usec = (double) INSTR_TIME_GET_MICROSEC(diff);
-
+			
+			/* should we aggregate the results or not? */
+			if (use_log_agg) {
+				
+				/* are we still in the same interval? if yes, accumulate the
+				 * values (print them otherwise) */
+				if (agg->start_time + naggseconds >= INSTR_TIME_GET_DOUBLE(now)) {
+					
+					/* accumulate */
+					agg->cnt += 1;
+					
+					agg->min_duration = (usec < agg->min_duration) ? usec : agg->min_duration;
+					agg->max_duration = (usec > agg->max_duration) ? usec : agg->max_duration;
+					
+					agg->sum  += usec;
+					agg->sum2 += usec * usec;
+					
+				} else {
+					
+					/* print */
 #ifndef WIN32
-			/* This is more than we really ought to know about instr_time */
-			fprintf(logfile, "%d %d %.0f %d %ld %ld\n",
-					st->id, st->cnt, usec, st->use_file,
-					(long) now.tv_sec, (long) now.tv_usec);
+					/* This is more than we really ought to know about instr_time */
+					fprintf(logfile, "%ld %d %.0f %.0f %.0f %.0f\n",
+								agg->start_time, agg->cnt, agg->sum, agg->sum2,
+								agg->min_duration, agg->max_duration);
 #else
-			/* On Windows, instr_time doesn't provide a timestamp anyway */
-			fprintf(logfile, "%d %d %.0f %d 0 0\n",
-					st->id, st->cnt, usec, st->use_file);
+					/* On Windows, instr_time doesn't provide a timestamp anyway */
+					// FIXME
+//					fprintf(logfile, "%d %ld %.0f %d 0 0\n",
+//							st->id, aggvals.start_time, aggvals.cnt, usec, st->use_file);
 #endif
+					/* and now reset the values (include the current) */
+					agg->cnt = 1;
+					agg->min_duration = usec;
+					agg->max_duration = usec;
+						
+					agg->sum = usec;
+					agg->sum2 = usec * usec;
+					
+					/* move to the next interval (there may be transactions longer than
+					 * the desired interval */
+					while (agg->start_time + naggseconds < INSTR_TIME_GET_DOUBLE(now)) {
+						agg->start_time = agg->start_time + naggseconds;
+					}
+				}
+				
+			} else {
+				/* no, print raw transactions */
+#ifndef WIN32
+				/* This is more than we really ought to know about instr_time */
+				fprintf(logfile, "%d %d %.0f %d %ld %ld\n",
+						st->id, st->cnt, usec, st->use_file,
+						(long) now.tv_sec, (long) now.tv_usec);
+#else
+				/* On Windows, instr_time doesn't provide a timestamp anyway */
+				fprintf(logfile, "%d %d %.0f %d 0 0\n",
+						st->id, st->cnt, usec, st->use_file);
+#endif
+			}
 		}
 
 		if (commands[st->state]->type == SQL_COMMAND)
@@ -1962,7 +2038,7 @@ main(int argc, char **argv)
 	state = (CState *) xmalloc(sizeof(CState));
 	memset(state, 0, sizeof(CState));
 
-	while ((c = getopt_long(argc, argv, "ih:nvp:dSNc:j:Crs:t:T:U:lf:D:F:M:", long_options, &optindex)) != -1)
+	while ((c = getopt_long(argc, argv, "ih:nvp:dSNc:j:Crs:t:T:U:lf:A:D:F:M:", long_options, &optindex)) != -1)
 	{
 		switch (c)
 		{
@@ -2070,6 +2146,15 @@ main(int argc, char **argv)
 			case 'l':
 				use_log = true;
 				break;
+			case 'A':
+				use_log_agg = true;
+				naggseconds = atoi(optarg);
+				if (naggseconds <= 0)
+				{
+					fprintf(stderr, "invalid number of seconds for aggregation: %d\n", naggseconds);
+					exit(1);
+				}
+				break;
 			case 'f':
 				ttype = 3;
 				filename = optarg;
@@ -2158,6 +2243,11 @@ main(int argc, char **argv)
 		exit(1);
 	}
 
+	if (naggseconds > duration) {
+		fprintf(stderr, "number of seconds for aggregation (%d) must not be higher that test duration (%d)\n", naggseconds, duration);
+		exit(1);
+	}
+
 	/*
 	 * is_latencies only works with multiple threads in thread-based
 	 * implementations, not fork-based ones, because it supposes that the
@@ -2416,7 +2506,9 @@ threadRun(void *arg)
 	int			nstate = thread->nstate;
 	int			remains = nstate;		/* number of remaining clients */
 	int			i;
-
+	
+	AggVals		aggs;
+	
 	result = xmalloc(sizeof(TResult));
 	INSTR_TIME_SET_ZERO(result->conn_time);
 
@@ -2452,6 +2544,8 @@ threadRun(void *arg)
 	INSTR_TIME_SET_CURRENT(result->conn_time);
 	INSTR_TIME_SUBTRACT(result->conn_time, thread->start_time);
 
+	agg_vals_init(&aggs, thread->start_time);
+	
 	/* send start up queries in async manner */
 	for (i = 0; i < nstate; i++)
 	{
@@ -2460,7 +2554,7 @@ threadRun(void *arg)
 		int			prev_ecnt = st->ecnt;
 
 		st->use_file = getrand(thread, 0, num_files - 1);
-		if (!doCustom(thread, st, &result->conn_time, logfile))
+		if (!doCustom(thread, st, &result->conn_time, logfile, &aggs))
 			remains--;			/* I've aborted */
 
 		if (st->ecnt > prev_ecnt && commands[st->state]->type == META_COMMAND)
@@ -2562,7 +2656,7 @@ threadRun(void *arg)
 			if (st->con && (FD_ISSET(PQsocket(st->con), &input_mask)
 							|| commands[st->state]->type == META_COMMAND))
 			{
-				if (!doCustom(thread, st, &result->conn_time, logfile))
+				if (!doCustom(thread, st, &result->conn_time, logfile, &aggs))
 					remains--;	/* I've aborted */
 			}
 
@@ -2589,7 +2683,6 @@ done:
 	return result;
 }
 
-
 /*
  * Support for duration option: set timer_exceeded after so many seconds.
  */
@@ -2783,4 +2876,4 @@ pthread_join(pthread_t th, void **thread_return)
 	return 0;
 }
 
-#endif   /* WIN32 */
+#endif   /* WIN32 */
\ No newline at end of file
diff --git a/doc/src/sgml/pgbench.sgml b/doc/src/sgml/pgbench.sgml
index 437fcea..8929b96 100644
--- a/doc/src/sgml/pgbench.sgml
+++ b/doc/src/sgml/pgbench.sgml
@@ -317,6 +317,18 @@ pgbench <optional> <replaceable>options</> </optional> <replaceable>dbname</>
      </varlistentry>
 
      <varlistentry>
+      <term><option>-A</option> <replaceable>seconds</></term>
+      <listitem>
+       <para>
+        Length of aggregation interval (in seconds). May be used only together
+        with <application>-l</application> - with this option, the log contains
+        per-interval summary (number of transactions, min/max latency and two
+        additional fields useful for variance estimation). Default is 1.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
       <term><option>-M</option> <replaceable>querymode</></term>
       <listitem>
        <para>
@@ -712,8 +724,9 @@ END;
   <title>Per-Transaction Logging</title>
 
   <para>
-   With the <option>-l</> option, <application>pgbench</> writes the time
-   taken by each transaction to a log file.  The log file will be named
+   With the <option>-l</> option, and without the <option>-A</option>,
+   <application>pgbench</> writes the time taken by each transaction
+   to a log file.  The log file will be named
    <filename>pgbench_log.<replaceable>nnn</></filename>, where
    <replaceable>nnn</> is the PID of the pgbench process.
    If the <option>-j</> option is 2 or higher, creating multiple worker
@@ -750,6 +763,37 @@ END;
  0 201 2513 0 1175850569 608
  0 202 2038 0 1175850569 2663
 </screen></para>
+
+  <title>Aggregated Logging</title>
+  
+  <para>
+   With the <option>-A</option> option, the logs use a bit different format:
+
+<synopsis>
+<replaceable>interval_start</> <replaceable>num_of_transactions</> <replaceable>latency_sum</> <replaceable>latency_2_sum</> <replaceable>min_latency</> <replaceable>max_latency</>
+</synopsis>
+
+   where <replaceable>interval_start</> is the start of the interval (UNIX epoch
+   format timestamp), <replaceable>num_of_transactions</> is the number of transactions
+   within the interval, <replaceable>latency_sum</replaceable> is a sum of latencies
+   (so you can compute average latency easily). The following two fields are useful
+   for variance estimation - <replaceable>latency_sum</> is a sum of latencies and
+   <replaceable>latency_2_sum</> is a sum of 2nd powers of latencies. The last two
+   fields are <replaceable>min_latency</> - a minimum latency within the interval, and
+   <replaceable>max_latency</> - maximum latency within the interval. A transaction is
+   counted into the interval when it was committed.
+  </para>
+
+  <para>
+   Here is example outputs:
+<screen>
+1345828501 5601 1542744 483552416 61 2573
+1345828503 7884 1979812 565806736 60 1479
+1345828505 7208 1979422 567277552 59 1391
+1345828507 7685 1980268 569784714 60 1398
+1345828509 7073 1979779 573489941 236 1411
+</screen></para>
+
  </refsect2>
 
  <refsect2>