Import wiredtiger: 7bf362af190a36a31589d3d78eb1cd1a5963b79d from branch mongodb-4.4

ref: 55d47c210c..7bf362af19
for: 4.4.0-rc7

WT-6056       Set DATA_CORRUPTION flag so we return TRY_SALVAGE on open
WT-6129       Add test to ensure prepare transactions will not lead to cache stuck
WT-6267       Fix not deleting prepared update when hs cursor is placed on another key

5 files changed, 239 insertions, 7 deletions

diff --git a/src/third_party/wiredtiger/import.data b/src/third_party/wiredtiger/import.data
index 34221ab1fa5..2af6d667182 100644
--- a/src/third_party/wiredtiger/import.data
+++ b/src/third_party/wiredtiger/import.data
@@ -2,5 +2,5 @@
     "vendor": "wiredtiger",
     "github": "wiredtiger/wiredtiger.git",
     "branch": "mongodb-4.4",
-    "commit": "55d47c210ccb07a115c2e6f71ee90367c7c3ab77"
+    "commit": "7bf362af190a36a31589d3d78eb1cd1a5963b79d"
 }
diff --git a/src/third_party/wiredtiger/src/meta/meta_turtle.c b/src/third_party/wiredtiger/src/meta/meta_turtle.c
index 119438fefcd..10605a71020 100644
--- a/src/third_party/wiredtiger/src/meta/meta_turtle.c
+++ b/src/third_party/wiredtiger/src/meta/meta_turtle.c
@@ -381,6 +381,7 @@ err:
      */
     if (ret == 0 || strcmp(key, WT_METADATA_COMPAT) == 0 || F_ISSET(S2C(session), WT_CONN_SALVAGE))
         return (ret);
+    F_SET(S2C(session), WT_CONN_DATA_CORRUPTION);
     WT_RET_PANIC(session, WT_TRY_SALVAGE, "%s: fatal turtle file read error", WT_METADATA_TURTLE);
 }
 
@@ -437,5 +438,6 @@ err:
      */
     if (ret == 0)
         return (ret);
+    F_SET(conn, WT_CONN_DATA_CORRUPTION);
     WT_RET_PANIC(session, ret, "%s: fatal turtle file update error", WT_METADATA_TURTLE);
 }
diff --git a/src/third_party/wiredtiger/src/txn/txn.c b/src/third_party/wiredtiger/src/txn/txn.c
index 769e54225a5..f20bfba99fb 100644
--- a/src/third_party/wiredtiger/src/txn/txn.c
+++ b/src/third_party/wiredtiger/src/txn/txn.c
@@ -631,17 +631,21 @@ __txn_append_hs_record(WT_SESSION_IMPL *session, WT_CURSOR *hs_cursor, WT_ITEM *
 
     WT_ERR(hs_cursor->get_key(hs_cursor, &hs_btree_id, hs_key, &hs_start_ts, &hs_counter));
 
-    /* Stop before crossing over to the next btree */
-    if (hs_btree_id != S2BT(session)->id)
+    /* Not found if we cross the tree boundary. */
+    if (hs_btree_id != S2BT(session)->id) {
+        ret = WT_NOTFOUND;
         goto done;
+    }
 
     /*
      * Keys are sorted in an order, skip the ones before the desired key, and bail out if we have
      * crossed over the desired key and not found the record we are looking for.
      */
     WT_ERR(__wt_compare(session, NULL, hs_key, key, &cmp));
-    if (cmp != 0)
+    if (cmp != 0) {
+        ret = WT_NOTFOUND;
         goto done;
+    }
 
     /*
      * As part of the history store search, we never get an exact match based on our search criteria
@@ -925,9 +929,11 @@ __txn_resolve_prepared_op(WT_SESSION_IMPL *session, WT_TXN_OP *op, bool commit,
           true);
 
         if (ret == 0)
-            WT_ERR(__txn_append_hs_record(session, hs_cursor, &op->u.op_row.key, cbt->ref->page,
-              upd, commit, &fix_upd, &upd_appended));
-        else if (ret == WT_NOTFOUND && !commit) {
+            /* Not found if we cross the tree or key boundary. */
+            WT_ERR_NOTFOUND_OK(__txn_append_hs_record(session, hs_cursor, &op->u.op_row.key,
+                                 cbt->ref->page, upd, commit, &fix_upd, &upd_appended),
+              true);
+        if (ret == WT_NOTFOUND && !commit) {
             /*
              * Allocate a tombstone so that when we reconcile the update chain we don't copy the
              * prepared cell, which is now associated with a rolled back prepare, and instead write
diff --git a/src/third_party/wiredtiger/test/csuite/wt4156_metadata_salvage/main.c b/src/third_party/wiredtiger/test/csuite/wt4156_metadata_salvage/main.c
index 0bc6adbbc00..aaa1b49f961 100644
--- a/src/third_party/wiredtiger/test/csuite/wt4156_metadata_salvage/main.c
+++ b/src/third_party/wiredtiger/test/csuite/wt4156_metadata_salvage/main.c
@@ -345,6 +345,8 @@ wt_open_corrupt(const char *sfx)
     WT_DECL_RET;
     char buf[1024];
 
+    /* The child should not abort the test in the message handler. Set it here, don't inherit. */
+    test_abort = false;
     if (sfx != NULL)
         testutil_check(__wt_snprintf(buf, sizeof(buf), "%s.%s", home, sfx));
     else
diff --git a/src/third_party/wiredtiger/test/suite/test_prepare_hs04.py b/src/third_party/wiredtiger/test/suite/test_prepare_hs04.py
new file mode 100644
index 00000000000..c5d3a9b7050
--- /dev/null
+++ b/src/third_party/wiredtiger/test/suite/test_prepare_hs04.py
@@ -0,0 +1,222 @@
+#!/usr/bin/env python
+#
+# Public Domain 2014-2020 MongoDB, Inc.
+# Public Domain 2008-2014 WiredTiger, Inc.
+#
+# This is free and unencumbered software released into the public domain.
+#
+# Anyone is free to copy, modify, publish, use, compile, sell, or
+# distribute this software, either in source code form or as a compiled
+# binary, for any purpose, commercial or non-commercial, and by any
+# means.
+#
+# In jurisdictions that recognize copyright laws, the author or authors
+# of this software dedicate any and all copyright interest in the
+# software to the public domain. We make this dedication for the benefit
+# of the public at large and to the detriment of our heirs and
+# successors. We intend this dedication to be an overt act of
+# relinquishment in perpetuity of all present and future rights to this
+# software under copyright law.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+# OTHER DEALINGS IN THE SOFTWARE.
+
+from helper import copy_wiredtiger_home
+import wiredtiger, wttest
+from wtdataset import SimpleDataSet
+from wtscenario import make_scenarios
+from wiredtiger import stat
+
+def timestamp_str(t):
+    return '%x' % t
+
+# test_prepare_hs04.py
+# Read prepared updates from on-disk with ignore_prepare.
+# Committing or aborting a prepared update when there exists a tombstone for that key already.
+#
+class test_prepare_hs04(wttest.WiredTigerTestCase):
+    # Force a small cache.
+    conn_config = 'cache_size=50MB,statistics=(fast)'
+
+    # Create a small table.
+    uri = "table:test_prepare_hs04"
+
+    nsessions = 3
+    nkeys = 40
+    nrows = 100
+
+    scenarios = make_scenarios([
+        ('commit_transaction', dict(commit=True)),
+        ('rollback_transaction', dict(commit=False))
+    ])
+
+    def get_stat(self, stat):
+        stat_cursor = self.session.open_cursor('statistics:')
+        val = stat_cursor[stat][2]
+        stat_cursor.close()
+        return val
+
+    def search_keys_timestamp_and_ignore(self, ds, txn_config, expected_value, conflict=False):
+        cursor = self.session.open_cursor(self.uri)
+
+        commit_key = "C"
+        self.session.begin_transaction(txn_config)
+        for i in range(1, self.nsessions * self.nkeys):
+            key = commit_key + ds.key(self.nrows + i)
+            cursor.set_key(key)
+            if conflict == True:
+                self.assertRaisesException(wiredtiger.WiredTigerError, lambda:cursor.search(), expected_value)
+            elif expected_value == None:
+                self.assertEqual(cursor.search(), wiredtiger.WT_NOTFOUND)
+            else:
+                self.assertEqual(cursor.search(), 0)
+                self.assertEqual(cursor.get_value(), expected_value)
+        cursor.close()
+        self.session.commit_transaction()
+
+    def prepare_updates(self, ds):
+
+        # Commit some updates to get eviction and history store fired up
+        # Insert a key at timestamp 1
+        commit_key = "C"
+        commit_value = b"bbbbb" * 100
+        cursor = self.session.open_cursor(self.uri)
+        for i in range(1, self.nsessions * self.nkeys):
+            self.session.begin_transaction('isolation=snapshot')
+            key = commit_key + ds.key(self.nrows + i)
+            cursor.set_key(key)
+            cursor.set_value(commit_value)
+            self.assertEquals(cursor.insert(), 0)
+            self.session.commit_transaction('commit_timestamp=' + timestamp_str(1))
+        cursor.close()
+
+        # Call checkpoint
+        self.session.checkpoint()
+
+        cursor = self.session.open_cursor(self.uri)
+        for i in range(1, self.nsessions * self.nkeys):
+            self.session.begin_transaction('isolation=snapshot')
+            key = commit_key + ds.key(self.nrows + i)
+            cursor.set_key(key)
+            self.assertEquals(cursor.remove(), 0)
+            self.session.commit_transaction('commit_timestamp=' + timestamp_str(10))
+        cursor.close()
+
+        hs_writes_start = self.get_stat(stat.conn.cache_write_hs)
+        # Have prepared updates in multiple sessions. This should ensure writing
+        # prepared updates to the data store
+        # Insert the same key at timestamp 20, but with prepare updates.
+        sessions = [0] * self.nsessions
+        cursors = [0] * self.nsessions
+        prepare_value = b"ccccc" * 100
+        for j in range (0, self.nsessions):
+            sessions[j] = self.conn.open_session()
+            sessions[j].begin_transaction('isolation=snapshot')
+            cursors[j] = sessions[j].open_cursor(self.uri)
+            # Each session will update many consecutive keys.
+            start = (j * self.nkeys)
+            end = start + self.nkeys
+            for i in range(start, end):
+                cursors[j].set_key(commit_key + ds.key(self.nrows + i))
+                cursors[j].set_value(prepare_value)
+                self.assertEquals(cursors[j].insert(), 0)
+            sessions[j].prepare_transaction('prepare_timestamp=' + timestamp_str(20))
+
+        hs_writes = self.get_stat(stat.conn.cache_write_hs) - hs_writes_start
+
+        # Assert if not writing anything to the history store.
+        self.assertGreaterEqual(hs_writes, 0)
+
+        txn_config = 'read_timestamp=' + timestamp_str(5) + ',ignore_prepare=true'
+        # Search keys with timestamp 5, ignore_prepare=true and expect the cursor search to return 0 (key found)
+        self.search_keys_timestamp_and_ignore(ds, txn_config, commit_value)
+
+        txn_config = 'read_timestamp=' + timestamp_str(20) + ',ignore_prepare=true'
+        # Search keys with timestamp 20, ignore_prepare=true, expect the cursor to return wiredtiger.WT_NOTFOUND
+        self.search_keys_timestamp_and_ignore(ds, txn_config, None)
+
+        prepare_conflict_msg = '/conflict with a prepared update/'
+        txn_config = 'read_timestamp=' + timestamp_str(20) + ',ignore_prepare=false'
+        # Search keys with timestamp 20, ignore_prepare=false and expect the cursor the cursor search to return prepare conflict message
+        self.search_keys_timestamp_and_ignore(ds, txn_config, prepare_conflict_msg, True)
+
+        # If commit is True then commit the transactions and simulate a crash which would
+        # eventualy rollback transactions.
+        if self.commit == True:
+            # Commit the prepared_transactions with timestamp 30.
+            for j in range (0, self.nsessions):
+                sessions[j].commit_transaction(
+                    'commit_timestamp=' + timestamp_str(30) + ',durable_timestamp=' + timestamp_str(30))
+
+        self.session.checkpoint()
+
+        # Simulate a crash by copying to a new directory(RESTART).
+        copy_wiredtiger_home(".", "RESTART")
+
+        # Open the new directory.
+        self.conn = self.setUpConnectionOpen("RESTART")
+        self.session = self.setUpSessionOpen(self.conn)
+
+        # After simulating a crash, search for the keys inserted.
+
+        txn_config = 'read_timestamp=' + timestamp_str(5) + ',ignore_prepare=false'
+        # Search keys with timestamp 5, ignore_prepare=false and expect the cursor value to be commit_value.
+        self.search_keys_timestamp_and_ignore(ds, txn_config, commit_value)
+
+        txn_config = 'read_timestamp=' + timestamp_str(20) + ',ignore_prepare=true'
+        # Search keys with timestamp 20, ignore_prepare=true and expect the cursor search to return WT_NOTFOUND.
+        self.search_keys_timestamp_and_ignore(ds, txn_config, None)
+
+        txn_config = 'read_timestamp=' + timestamp_str(20) + ',ignore_prepare=false'
+        # Search keys with timestamp 20, ignore_prepare=false and expect the cursor search to return WT_NOTFOUND.
+        self.search_keys_timestamp_and_ignore(ds, txn_config, None)
+
+        # If commit is true then the commit_tramsactions was called and we will expect prepare_value.
+        if self.commit == True:
+            txn_config = 'read_timestamp=' + timestamp_str(30) + ',ignore_prepare=true'
+            # Search keys with timestamp 30, ignore_prepare=true and expect the cursor value to be prepare_value.
+            self.search_keys_timestamp_and_ignore(ds, txn_config, prepare_value)
+        else:
+            # Commit is false and we simulated a crash/restart which would have rolled-back the transactions, therefore we expect the
+            # cursor search to return WT_NOTFOUND.
+            txn_config = 'read_timestamp=' + timestamp_str(30) + ',ignore_prepare=true'
+            # Search keys with timestamp 30, ignore_prepare=true and expect the cursor value to return WT_NOTFOUND.
+            self.search_keys_timestamp_and_ignore(ds, txn_config, None)
+
+        if self.commit == True:
+            txn_config = 'read_timestamp=' + timestamp_str(30) + ',ignore_prepare=false'
+            # Search keys with timestamp 30, ignore_prepare=false and expect the cursor value to be prepare_value.
+            self.search_keys_timestamp_and_ignore(ds, txn_config, prepare_value)
+        else:
+            # Commit is false and we simulated a crash/restart which would have rolled-back the transactions, therefore we expect the
+            # cursor search to return WT_NOTFOUND.
+            txn_config = 'read_timestamp=' + timestamp_str(30) + ',ignore_prepare=false'
+            # Search keys with timestamp 30, ignore_prepare=false and expect the cursor value to return WT_NOTFOUND.
+            self.search_keys_timestamp_and_ignore(ds, txn_config, None)
+
+    def test_prepare_hs(self):
+
+        ds = SimpleDataSet(self, self.uri, self.nrows, key_format="S", value_format='u')
+        ds.populate()
+        bigvalue = b"aaaaa" * 100
+
+        # Initially load huge data
+        cursor = self.session.open_cursor(self.uri)
+        for i in range(1, 10000):
+            cursor.set_key(ds.key(self.nrows + i))
+            cursor.set_value(bigvalue)
+            self.assertEquals(cursor.insert(), 0)
+        cursor.close()
+        self.session.checkpoint()
+
+        # We put prepared updates in multiple sessions so that we do not hang
+        # because of cache being full with uncommitted updates.
+        self.prepare_updates(ds)
+
+if __name__ == '__main__':
+    wttest.run()