WT-3489 Add an API to set the tracked commit_timestamp. (#3562)

This is required during rollback, because the commit_timestamp
determines how much of the oplog is visible.

6 files changed, 139 insertions, 83 deletions

diff --git a/dist/api_data.py b/dist/api_data.py
index 3a0822b33b2..ee2f14b980b 100644
--- a/dist/api_data.py
+++ b/dist/api_data.py
@@ -1252,6 +1252,13 @@ methods = {
 ]),
 
 'WT_CONNECTION.set_timestamp' : Method([
+    Config('commit_timestamp', '', r'''
+        reset the maximum commit timestamp tracked by WiredTiger.  This will
+        cause future calls to WT_CONNECTION::query_timestamp to ignore commit
+        timestamps greater than the specified value until the next commit moves
+        the tracked commit timestamp forwards.  This is only intended for use
+        where the application is rolling back locally committed transactions.
+        See @ref transaction_timestamps'''),
     Config('oldest_timestamp', '', r'''
         future commits and queries will be no earlier than the specified
         timestamp. Supplied values must be monotonically increasing.
diff --git a/examples/c/ex_all.c b/examples/c/ex_all.c
index 672c54fa1f6..02237faf4e9 100644
--- a/examples/c/ex_all.c
+++ b/examples/c/ex_all.c
@@ -863,6 +863,10 @@ transaction_ops(WT_CONNECTION *conn, WT_SESSION *session)
 	/*! [query timestamp] */
 	}
 
+	/*! [set commit timestamp] */
+	error_check(conn->set_timestamp(conn, "commit_timestamp=2a"));
+	/*! [set commit timestamp] */
+
 	/*! [set oldest timestamp] */
 	error_check(conn->set_timestamp(conn, "oldest_timestamp=2a"));
 	/*! [set oldest timestamp] */
diff --git a/src/config/config_def.c b/src/config/config_def.c
index f0c4e47e7e7..c53a63ccb25 100644
--- a/src/config/config_def.c
+++ b/src/config/config_def.c
@@ -179,6 +179,7 @@ static const WT_CONFIG_CHECK confchk_WT_CONNECTION_reconfigure[] = {
 };
 
 static const WT_CONFIG_CHECK confchk_WT_CONNECTION_set_timestamp[] = {
+	{ "commit_timestamp", "string", NULL, NULL, NULL, 0 },
 	{ "oldest_timestamp", "string", NULL, NULL, NULL, 0 },
 	{ "stable_timestamp", "string", NULL, NULL, NULL, 0 },
 	{ NULL, NULL, NULL, NULL, NULL, 0 }
@@ -1152,8 +1153,8 @@ static const WT_CONFIG_ENTRY config_entries[] = {
 	  NULL, 0
 	},
 	{ "WT_CONNECTION.set_timestamp",
-	  "oldest_timestamp=,stable_timestamp=",
-	  confchk_WT_CONNECTION_set_timestamp, 2
+	  "commit_timestamp=,oldest_timestamp=,stable_timestamp=",
+	  confchk_WT_CONNECTION_set_timestamp, 3
 	},
 	{ "WT_CURSOR.close",
 	  "",
diff --git a/src/include/wiredtiger.in b/src/include/wiredtiger.in
index ef863176ba8..5d087447c5a 100644
--- a/src/include/wiredtiger.in
+++ b/src/include/wiredtiger.in
@@ -2254,10 +2254,19 @@ struct __wt_connection {
 	/*!
 	 * Set a global transaction timestamp.
 	 *
+	 * @snippet ex_all.c set commit timestamp
+	 *
 	 * @snippet ex_all.c set oldest timestamp
 	 *
 	 * @param connection the connection handle
 	 * @configstart{WT_CONNECTION.set_timestamp, see dist/api_data.py}
+	 * @config{commit_timestamp, reset the maximum commit timestamp tracked
+	 * by WiredTiger.  This will cause future calls to
+	 * WT_CONNECTION::query_timestamp to ignore commit timestamps greater
+	 * than the specified value until the next commit moves the tracked
+	 * commit timestamp forwards.  This is only intended for use where the
+	 * application is rolling back locally committed transactions.  See @ref
+	 * transaction_timestamps., a string; default empty.}
 	 * @config{oldest_timestamp, future commits and queries will be no
 	 * earlier than the specified timestamp.  Supplied values must be
 	 * monotonically increasing.  See @ref transaction_timestamps., a
diff --git a/src/txn/txn_timestamp.c b/src/txn/txn_timestamp.c
index 4caf0102e3c..e67f43a750f 100644
--- a/src/txn/txn_timestamp.c
+++ b/src/txn/txn_timestamp.c
@@ -290,93 +290,119 @@ __wt_txn_update_pinned_timestamp(WT_SESSION_IMPL *session)
 int
 __wt_txn_global_set_timestamp(WT_SESSION_IMPL *session, const char *cfg[])
 {
-	WT_CONFIG_ITEM oldest_cval, stable_cval;
-	bool has_oldest, has_stable;
+	WT_CONFIG_ITEM commit_cval, oldest_cval, stable_cval;
+	bool has_commit, has_oldest, has_stable;
+
+	WT_RET(__wt_config_gets_def(session,
+	    cfg, "commit_timestamp", 0, &commit_cval));
+	has_commit = commit_cval.len != 0;
 
-	/*
-	 * Look for a commit timestamp.
-	 */
 	WT_RET(__wt_config_gets_def(session,
 	    cfg, "oldest_timestamp", 0, &oldest_cval));
+	has_oldest = oldest_cval.len != 0;
+
 	WT_RET(__wt_config_gets_def(session,
 	    cfg, "stable_timestamp", 0, &stable_cval));
-	if (oldest_cval.len != 0)
-		has_oldest = true;
-	else
-		has_oldest = false;
-	if (stable_cval.len != 0)
-		has_stable = true;
-	else
-		has_stable = false;
-	if (has_oldest || has_stable) {
+	has_stable = stable_cval.len != 0;
+
+	/* If no timestamp was supplied, there's nothing to do. */
+	if (!has_commit && !has_oldest && !has_stable)
+		return (0);
+
 #ifdef HAVE_TIMESTAMPS
-		WT_TXN_GLOBAL *txn_global;
-		wt_timestamp_t oldest_ts, stable_ts;
-
-		txn_global = &S2C(session)->txn_global;
-		/*
-		 * Parsing will initialize the timestamp to zero even if
-		 * it is not configured.
-		 */
-		WT_RET(__wt_txn_parse_timestamp(
-		    session, "oldest", &oldest_ts, &oldest_cval));
-		WT_RET(__wt_txn_parse_timestamp(
-		    session, "stable", &stable_ts, &stable_cval));
-		__wt_writelock(session, &txn_global->rwlock);
-		/*
-		 * First do error checking on the timestamp values.  The
-		 * oldest timestamp must always be less than or equal to
-		 * the stable timestamp.  If we're only setting one
-		 * then compare against the system timestamp.  If we're
-		 * setting both then compare the passed in values.
-		 */
-		if ((has_oldest && !has_stable &&	/* only oldest given */
-		    txn_global->has_stable_timestamp &&
-		    __wt_timestamp_cmp(&oldest_ts,
-		    &txn_global->stable_timestamp) > 0) ||
-		    (has_stable && !has_oldest &&	/* only stable given */
-		    txn_global->has_oldest_timestamp &&
-		    __wt_timestamp_cmp(&stable_ts,
-		    &txn_global->oldest_timestamp) < 0) ||
-		    (has_oldest && has_stable &&	/* both given */
-		    __wt_timestamp_cmp(&oldest_ts, &stable_ts) > 0)) {
-			__wt_writeunlock(session, &txn_global->rwlock);
-			WT_RET_MSG(session, EINVAL,
-			    "set_timestamp: oldest timestamp must not be "
-			    "later than stable timestamp");
-		}
-		if (has_oldest) {
-			/*
-			 * This method can be called from multiple threads,
-			 * check that we are moving the global oldest
-			 * timestamp forwards.
-			 */
-			if (!txn_global->has_oldest_timestamp ||
-			    __wt_timestamp_cmp(&txn_global->oldest_timestamp,
-			    &oldest_ts) < 0) {
-				__wt_timestamp_set(
-				    &txn_global->oldest_timestamp, &oldest_ts);
-				txn_global->has_oldest_timestamp = true;
-				txn_global->oldest_is_pinned = false;
-			}
-		}
-		if (has_stable) {
-			/*
-			 * This method can be called from multiple threads,
-			 * check that we are moving the global stable
-			 * timestamp forwards.
-			 */
-			if (!txn_global->has_stable_timestamp ||
-			    __wt_timestamp_cmp(&txn_global->stable_timestamp,
-			    &stable_ts) < 0) {
-				__wt_timestamp_set(
-				    &txn_global->stable_timestamp, &stable_ts);
-				txn_global->has_stable_timestamp = true;
-				txn_global->stable_is_pinned = false;
-			}
-		}
+	{
+	WT_TXN_GLOBAL *txn_global;
+	wt_timestamp_t commit_ts, oldest_ts, stable_ts;
+
+	txn_global = &S2C(session)->txn_global;
+	/*
+	 * Parsing will initialize the timestamp to zero even if
+	 * it is not configured.
+	 */
+	WT_RET(__wt_txn_parse_timestamp(
+	    session, "commit", &commit_ts, &commit_cval));
+	WT_RET(__wt_txn_parse_timestamp(
+	    session, "oldest", &oldest_ts, &oldest_cval));
+	WT_RET(__wt_txn_parse_timestamp(
+	    session, "stable", &stable_ts, &stable_cval));
+	__wt_writelock(session, &txn_global->rwlock);
+
+	/*
+	 * First do error checking on the timestamp values.  The
+	 * oldest timestamp must always be less than or equal to
+	 * the stable timestamp.  If we're only setting one
+	 * then compare against the system timestamp.  If we're
+	 * setting both then compare the passed in values.
+	 */
+	if (!has_commit && txn_global->has_commit_timestamp)
+		__wt_timestamp_set(&commit_ts, &txn_global->commit_timestamp);
+	if (!has_oldest && txn_global->has_oldest_timestamp)
+		__wt_timestamp_set(&oldest_ts, &txn_global->oldest_timestamp);
+	if (!has_stable && txn_global->has_oldest_timestamp)
+		__wt_timestamp_set(&stable_ts, &txn_global->stable_timestamp);
+
+	if ((has_commit || txn_global->has_commit_timestamp) &&
+	    (has_oldest || txn_global->has_oldest_timestamp) &&
+	    __wt_timestamp_cmp(&oldest_ts, &commit_ts) > 0) {
+		__wt_writeunlock(session, &txn_global->rwlock);
+		WT_RET_MSG(session, EINVAL,
+		    "set_timestamp: oldest timestamp must not be later than "
+		    "commit timestamp");
+	}
+
+	if ((has_commit || txn_global->has_commit_timestamp) &&
+	    (has_stable || txn_global->has_stable_timestamp) &&
+	    __wt_timestamp_cmp(&stable_ts, &commit_ts) > 0) {
 		__wt_writeunlock(session, &txn_global->rwlock);
+		WT_RET_MSG(session, EINVAL,
+		    "set_timestamp: stable timestamp must not be later than "
+		    "commit timestamp");
+	}
+
+	if ((has_oldest || txn_global->has_oldest_timestamp) &&
+	    (has_stable || txn_global->has_stable_timestamp) &&
+	    __wt_timestamp_cmp(&oldest_ts, &stable_ts) > 0) {
+		__wt_writeunlock(session, &txn_global->rwlock);
+		WT_RET_MSG(session, EINVAL,
+		    "set_timestamp: oldest timestamp must be later than "
+		    "stable timestamp");
+	}
+
+	/*
+	 * This method can be called from multiple threads, check that we are
+	 * moving the global timestamps forwards.
+	 *
+	 * The exception is the commit timestamp, where the application can
+	 * move it backwards (in fact, it only really makes sense to explicitly
+	 * move it backwards because it otherwise tracks the largest
+	 * commit_timestamp so it moves forward whenever transactions are
+	 * assigned timestamps).
+	 */
+	if (has_commit) {
+		__wt_timestamp_set(&txn_global->commit_timestamp, &commit_ts);
+		txn_global->has_commit_timestamp = true;
+	}
+
+	if (has_oldest && (!txn_global->has_oldest_timestamp ||
+	    __wt_timestamp_cmp(
+	    &oldest_ts, &txn_global->oldest_timestamp) > 0)) {
+		__wt_timestamp_set(&txn_global->oldest_timestamp, &oldest_ts);
+		txn_global->has_oldest_timestamp = true;
+		txn_global->oldest_is_pinned = false;
+	}
+
+	if (has_stable && (!txn_global->has_stable_timestamp ||
+	    __wt_timestamp_cmp(
+	    &stable_ts, &txn_global->stable_timestamp) > 0)) {
+		__wt_timestamp_set(&txn_global->stable_timestamp, &stable_ts);
+		txn_global->has_stable_timestamp = true;
+		txn_global->stable_is_pinned = false;
+	}
+	__wt_writeunlock(session, &txn_global->rwlock);
+
+	if (has_oldest || has_stable)
 		WT_RET(__wt_txn_update_pinned_timestamp(session));
+
 #else
 		WT_RET_MSG(session, EINVAL, "set_timestamp requires a "
 		    "version of WiredTiger built with timestamp support");
diff --git a/test/suite/test_timestamp02.py b/test/suite/test_timestamp02.py
index 735e954fc7f..e0d31ecbebb 100644
--- a/test/suite/test_timestamp02.py
+++ b/test/suite/test_timestamp02.py
@@ -97,8 +97,10 @@ class test_timestamp02(wttest.WiredTigerTestCase, suite_subprocess):
             self.check(self.session, 'read_timestamp=' + timestamp_str(t),
                 dict((k, 1) for k in orig_keys[:i+1]))
 
-        # Bump the oldest timestamp, we're not going back...
+        # Everything up to and including timestamp 100 has been committed.
         self.assertEqual(self.conn.query_timestamp(), timestamp_ret_str(100))
+
+        # Bump the oldest timestamp, we're not going back...
         self.conn.set_timestamp('oldest_timestamp=' + timestamp_str(100))
 
         # Update them and retry.
@@ -108,6 +110,14 @@ class test_timestamp02(wttest.WiredTigerTestCase, suite_subprocess):
             c[k] = 2
             self.session.commit_transaction('commit_timestamp=' + timestamp_str(k + 100))
 
+        # Everything up to and including timestamp 200 has been committed.
+        self.assertEqual(self.conn.query_timestamp(), timestamp_ret_str(200))
+
+        # Test that we can manually move the commit timestamp back
+        self.conn.set_timestamp('commit_timestamp=' + timestamp_str(150))
+        self.assertEqual(self.conn.query_timestamp(), timestamp_ret_str(150))
+        self.conn.set_timestamp('commit_timestamp=' + timestamp_str(200))
+
         # Now the stable timestamp before we read.
         self.conn.set_timestamp('stable_timestamp=' + timestamp_str(200))
 
@@ -116,7 +126,6 @@ class test_timestamp02(wttest.WiredTigerTestCase, suite_subprocess):
                 dict((k, (2 if j <= i else 1)) for j, k in enumerate(orig_keys)))
 
         # Bump the oldest timestamp, we're not going back...
-        self.assertEqual(self.conn.query_timestamp(), timestamp_ret_str(200))
         self.conn.set_timestamp('oldest_timestamp=' + timestamp_str(200))
 
         # Remove them and retry