Import wiredtiger: 8de74488f2bb2b5cba0404c345f568a2f72478d3 from branch mongodb-4.2r4.2.7-rc1r4.2.7

ref: 5de95caf8b..8de74488f2
for: 4.2.7

WT-5242       Minimize checkpoints pinned during backup
WT-6118       Fix missing checkpoint in backup
WT-6136       Record incremental extent lists before merging them with earlier checkpoints
WT-6137       Fix calculation of bits versus bytes for incremental bitmap
WT-6141       Disable checkpoint deletion during backup
WT-6156       Enable format to select "backup.incremental=log"

22 files changed, 656 insertions, 166 deletions

diff --git a/src/third_party/wiredtiger/dist/api_data.py b/src/third_party/wiredtiger/dist/api_data.py
index 45cca0ef829..a5df6227eee 100644
--- a/src/third_party/wiredtiger/dist/api_data.py
+++ b/src/third_party/wiredtiger/dist/api_data.py
@@ -1495,8 +1495,9 @@ methods = {
         including the named checkpoint, or
         \c "to=<checkpoint>" to drop all checkpoints before and
         including the named checkpoint.  Checkpoints cannot be
-        dropped while a hot backup is in progress or if open in
-        a cursor''', type='list'),
+        dropped if open in a cursor.  While a hot backup is in
+        progress, checkpoints created prior to the start of the
+        backup cannot be dropped''', type='list'),
     Config('force', 'false', r'''
         if false (the default), checkpoints may be skipped if the underlying object has not been
         modified, if true, this option forces the checkpoint''',
diff --git a/src/third_party/wiredtiger/import.data b/src/third_party/wiredtiger/import.data
index 4afe42eccce..28314b55190 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.2",
-    "commit": "5de95caf8b4514906b38e10e2063592907a5b3e6"
+    "commit": "8de74488f2bb2b5cba0404c345f568a2f72478d3"
 }
diff --git a/src/third_party/wiredtiger/src/block/block_ckpt.c b/src/third_party/wiredtiger/src/block/block_ckpt.c
index 6cbe80a9317..1a148606741 100644
--- a/src/third_party/wiredtiger/src/block/block_ckpt.c
+++ b/src/third_party/wiredtiger/src/block/block_ckpt.c
@@ -343,6 +343,119 @@ __ckpt_verify(WT_SESSION_IMPL *session, WT_CKPT *ckptbase)
 #endif
 
 /*
+ * __ckpt_add_blkmod_entry --
+ *     Add an offset/length entry to the bitstring based on granularity.
+ */
+static int
+__ckpt_add_blkmod_entry(
+  WT_SESSION_IMPL *session, WT_BLOCK_MODS *blk_mod, wt_off_t offset, wt_off_t len)
+{
+    uint64_t end_bit, start_bit;
+    uint32_t end_buf_bytes, end_rdup_bits, end_rdup_bytes;
+
+    WT_ASSERT(session, blk_mod->granularity != 0);
+    /*
+     * Figure out how the starting and ending bits based on the granularity and our offset and
+     * length.
+     */
+    start_bit = (uint64_t)offset / blk_mod->granularity;
+    end_bit = (uint64_t)(offset + len - 1) / blk_mod->granularity;
+    WT_ASSERT(session, end_bit < UINT32_MAX);
+    /* We want to grow the bitmap by 64 bits, or 8 bytes at a time. */
+    end_rdup_bits = WT_MAX(__wt_rduppo2((uint32_t)end_bit, 64), WT_BLOCK_MODS_LIST_MIN);
+    end_rdup_bytes = end_rdup_bits >> 3;
+    end_buf_bytes = (uint32_t)blk_mod->nbits >> 3;
+    /*
+     * We are doing a lot of shifting. Make sure that the number of bytes we end up with is a
+     * multiple of eight. We guarantee that in the rounding up call, but also make sure that the
+     * constant stays a multiple of eight.
+     */
+    WT_ASSERT(session, end_rdup_bytes % 8 == 0);
+    if (end_rdup_bytes > end_buf_bytes) {
+        /* If we don't have enough, extend the buffer. */
+        if (blk_mod->nbits == 0) {
+            WT_RET(__wt_buf_initsize(session, &blk_mod->bitstring, end_rdup_bytes));
+            memset(blk_mod->bitstring.mem, 0, end_rdup_bytes);
+        } else {
+            WT_RET(
+              __wt_buf_set(session, &blk_mod->bitstring, blk_mod->bitstring.data, end_rdup_bytes));
+            memset(
+              (uint8_t *)blk_mod->bitstring.mem + end_buf_bytes, 0, end_rdup_bytes - end_buf_bytes);
+        }
+        blk_mod->nbits = end_rdup_bits;
+    }
+    /* Set all the bits needed to record this offset/length pair. */
+    __bit_nset(blk_mod->bitstring.mem, start_bit, end_bit);
+    return (0);
+}
+
+/*
+ * __ckpt_add_blk_mods_alloc --
+ *     Add the checkpoint's allocated blocks to all valid incremental backup source identifiers.
+ */
+static int
+__ckpt_add_blk_mods_alloc(WT_SESSION_IMPL *session, WT_CKPT *ckptbase, WT_BLOCK_CKPT *ci)
+{
+    WT_BLOCK_MODS *blk_mod;
+    WT_CKPT *ckpt;
+    WT_EXT *ext;
+    u_int i;
+
+    WT_CKPT_FOREACH (ckptbase, ckpt) {
+        if (F_ISSET(ckpt, WT_CKPT_ADD))
+            break;
+    }
+    /* If this is not the live checkpoint or we don't care about incremental blocks, we're done. */
+    if (ckpt == NULL || !F_ISSET(ckpt, WT_CKPT_BLOCK_MODS))
+        return (0);
+    for (i = 0; i < WT_BLKINCR_MAX; ++i) {
+        blk_mod = &ckpt->backup_blocks[i];
+        /* If there is no information at this entry, we're done. */
+        if (!F_ISSET(blk_mod, WT_BLOCK_MODS_VALID))
+            continue;
+
+        WT_EXT_FOREACH (ext, ci->alloc.off) {
+            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ext->off, ext->size));
+        }
+    }
+    return (0);
+}
+
+/*
+ * __ckpt_add_blk_mods_ext --
+ *     Add a set of extent blocks to all valid incremental backup source identifiers.
+ */
+static int
+__ckpt_add_blk_mods_ext(WT_SESSION_IMPL *session, WT_CKPT *ckptbase, WT_BLOCK_CKPT *ci)
+{
+    WT_BLOCK_MODS *blk_mod;
+    WT_CKPT *ckpt;
+    u_int i;
+
+    WT_CKPT_FOREACH (ckptbase, ckpt) {
+        if (F_ISSET(ckpt, WT_CKPT_ADD))
+            break;
+    }
+    /* If this is not the live checkpoint or we don't care about incremental blocks, we're done. */
+    if (ckpt == NULL || !F_ISSET(ckpt, WT_CKPT_BLOCK_MODS))
+        return (0);
+    for (i = 0; i < WT_BLKINCR_MAX; ++i) {
+        blk_mod = &ckpt->backup_blocks[i];
+        /* If there is no information at this entry, we're done. */
+        if (!F_ISSET(blk_mod, WT_BLOCK_MODS_VALID))
+            continue;
+
+        if (ci->alloc.offset != WT_BLOCK_INVALID_OFFSET)
+            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ci->alloc.offset, ci->alloc.size));
+        if (ci->discard.offset != WT_BLOCK_INVALID_OFFSET)
+            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ci->discard.offset, ci->discard.size));
+        if (ci->avail.offset != WT_BLOCK_INVALID_OFFSET)
+            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ci->avail.offset, ci->avail.size));
+    }
+    return (0);
+}
+
+/*
  * __ckpt_process --
  *     Process the list of checkpoints.
  */
@@ -475,6 +588,12 @@ __ckpt_process(WT_SESSION_IMPL *session, WT_BLOCK *block, WT_CKPT *ckptbase)
     ckpt_size += ci->alloc.bytes;
     ckpt_size -= ci->discard.bytes;
 
+    /*
+     * Record the checkpoint's allocated blocks. Do so before skipping any processing and before
+     * possibly merging in blocks from any previous checkpoint.
+     */
+    WT_ERR(__ckpt_add_blk_mods_alloc(session, ckptbase, ci));
+
     /* Skip the additional processing if we aren't deleting checkpoints. */
     if (!deleting)
         goto live_update;
@@ -581,11 +700,9 @@ live_update:
         if (F_ISSET(ckpt, WT_CKPT_ADD)) {
             /*
              * !!!
-             * Our caller wants the final checkpoint size.  Setting
-             * the size here violates layering, but the alternative
-             * is a call for the btree layer to crack the checkpoint
-             * cookie into its components, and that's a fair amount
-             * of work.
+             * Our caller wants the final checkpoint size. Setting the size here violates layering,
+             * but the alternative is a call for the btree layer to crack the checkpoint cookie into
+             * its components, and that's a fair amount of work.
              */
             ckpt->size = ckpt_size;
 
@@ -654,78 +771,6 @@ err:
 }
 
 /*
- * __ckpt_add_blkmod_entry --
- *     Add an offset/length entry to the bitstring based on granularity.
- */
-static int
-__ckpt_add_blkmod_entry(
-  WT_SESSION_IMPL *session, WT_BLOCK_MODS *blk_mod, wt_off_t offset, wt_off_t len)
-{
-    uint64_t end, start;
-    uint32_t end_buf_bytes, end_rdup_bytes;
-
-    WT_ASSERT(session, blk_mod->granularity != 0);
-    start = (uint64_t)offset / blk_mod->granularity;
-    end = (uint64_t)(offset + len) / blk_mod->granularity;
-    WT_ASSERT(session, end < UINT32_MAX);
-    end_rdup_bytes = WT_MAX(__wt_rduppo2((uint32_t)end, 8), WT_BLOCK_MODS_LIST_MIN);
-    end_buf_bytes = (uint32_t)blk_mod->nbits >> 3;
-    /*
-     * We are doing a lot of shifting. Make sure that the number of bytes we end up with is a
-     * multiple of eight. We guarantee that in the rounding up call, but also make sure that the
-     * constant stays a multiple of eight.
-     */
-    WT_ASSERT(session, end_rdup_bytes % 8 == 0);
-    if (end_rdup_bytes > end_buf_bytes) {
-        /* If we don't have enough, extend the buffer. */
-        if (blk_mod->nbits == 0) {
-            WT_RET(__wt_buf_initsize(session, &blk_mod->bitstring, end_rdup_bytes));
-            memset(blk_mod->bitstring.mem, 0, end_rdup_bytes);
-        } else {
-            WT_RET(
-              __wt_buf_set(session, &blk_mod->bitstring, blk_mod->bitstring.data, end_rdup_bytes));
-            memset(
-              (uint8_t *)blk_mod->bitstring.mem + end_buf_bytes, 0, end_rdup_bytes - end_buf_bytes);
-        }
-        blk_mod->nbits = end_rdup_bytes << 3;
-    }
-
-    /* Set all the bits needed to record this offset/length pair. */
-    __bit_nset(blk_mod->bitstring.mem, start, end);
-    return (0);
-}
-
-/*
- * __ckpt_add_blk_mods --
- *     Add the blocks to all valid incremental backup source identifiers.
- */
-static int
-__ckpt_add_blk_mods(WT_SESSION_IMPL *session, WT_CKPT *ckpt, WT_BLOCK_CKPT *ci)
-{
-    WT_BLOCK_MODS *blk_mod;
-    WT_EXT *ext;
-    u_int i;
-
-    for (i = 0; i < WT_BLKINCR_MAX; ++i) {
-        blk_mod = &ckpt->backup_blocks[i];
-        /* If there is no information at this entry, we're done. */
-        if (!F_ISSET(blk_mod, WT_BLOCK_MODS_VALID))
-            continue;
-
-        WT_EXT_FOREACH (ext, ci->alloc.off)
-            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ext->off, ext->size));
-
-        if (ci->alloc.offset != WT_BLOCK_INVALID_OFFSET)
-            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ci->alloc.offset, ci->alloc.size));
-        if (ci->discard.offset != WT_BLOCK_INVALID_OFFSET)
-            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ci->discard.offset, ci->discard.size));
-        if (ci->avail.offset != WT_BLOCK_INVALID_OFFSET)
-            WT_RET(__ckpt_add_blkmod_entry(session, blk_mod, ci->avail.offset, ci->avail.size));
-    }
-    return (0);
-}
-
-/*
  * __ckpt_update --
  *     Update a checkpoint.
  */
@@ -747,9 +792,8 @@ __ckpt_update(
     WT_RET(__wt_block_extlist_check(session, &ci->alloc, &ci->discard));
 #endif
     /*
-     * Write the checkpoint's alloc and discard extent lists. After each write, remove any allocated
-     * blocks from the system's allocation list, checkpoint extent blocks don't appear on any extent
-     * lists.
+     * Write the checkpoint's alloc and discard extent lists. Note these blocks never appear on the
+     * system's allocation list, checkpoint extent blocks don't appear on any extent lists.
      */
     WT_RET(__wt_block_extlist_write(session, block, &ci->alloc, NULL));
     WT_RET(__wt_block_extlist_write(session, block, &ci->discard, NULL));
@@ -798,29 +842,34 @@ __ckpt_update(
     }
 
     /*
-     * If this is the live system, we need to record the list of blocks written for this checkpoint
-     * (including the blocks we allocated to write the extent lists).
+     * Record the blocks allocated to write the extent lists. We must record blocks in the live
+     * system's extent lists, as those blocks are a necessary part of the checkpoint a hot backup
+     * might recover. Update blocks in extent lists used to rewrite other checkpoints (for example,
+     * an intermediate checkpoint rewritten because a checkpoint was rolled into it), even though
+     * it's not necessary: those blocks aren't the last checkpoint in the file and so aren't
+     * included in a recoverable checkpoint, they don't matter on a hot backup target until they're
+     * allocated and * used in the context of a live system. Regardless, they shouldn't materially
+     * affect how much data we're writing, and it keeps things more consistent on the target to
+     * update them. (Ignore the live system's ckpt_avail list here. The blocks on that list were
+     * written into the final avail extent list which will be copied to the hot backup, and that's
+     * all that matters.)
      */
-    if (F_ISSET(ckpt, WT_CKPT_BLOCK_MODS))
-        WT_RET(__ckpt_add_blk_mods(session, ckpt, ci));
+    WT_RET(__ckpt_add_blk_mods_ext(session, ckptbase, ci));
 
     /*
      * Set the file size for the live system.
      *
      * !!!
-     * We do NOT set the file size when re-writing checkpoints because we
-     * want to test the checkpoint's blocks against a reasonable maximum
-     * file size during verification.  This is bad: imagine a checkpoint
-     * appearing early in the file, re-written, and then the checkpoint
-     * requires blocks at the end of the file, blocks after the listed file
-     * size.  If the application opens that checkpoint for writing
-     * (discarding subsequent checkpoints), we would truncate the file to
-     * the early chunk, discarding the re-written checkpoint information.
-     * The alternative, updating the file size has its own problems, in
-     * that case we'd work correctly, but we'd lose all of the blocks
-     * between the original checkpoint and the re-written checkpoint.
-     * Currently, there's no API to roll-forward intermediate checkpoints,
-     * if there ever is, this will need to be fixed.
+     * We do NOT set the file size when re-writing checkpoints because we want to test the
+     * checkpoint's blocks against a reasonable maximum file size during verification. This is bad:
+     * imagine a checkpoint appearing early in the file, re-written, and then the checkpoint
+     * requires blocks at the end of the file, blocks after the listed file size. If the application
+     * opens that checkpoint for writing (discarding subsequent checkpoints), we would truncate the
+     * file to the early chunk, discarding the re-written checkpoint information. The alternative,
+     * updating the file size has its own problems, in that case we'd work correctly, but we'd lose
+     * all of the blocks between the original checkpoint and the re-written checkpoint. Currently,
+     * there's no API to roll-forward intermediate checkpoints, if there ever is, this will need to
+     * be fixed.
      */
     if (is_live)
         ci->file_size = block->size;
diff --git a/src/third_party/wiredtiger/src/block/block_write.c b/src/third_party/wiredtiger/src/block/block_write.c
index d69371ee533..a8a0091d854 100644
--- a/src/third_party/wiredtiger/src/block/block_write.c
+++ b/src/third_party/wiredtiger/src/block/block_write.c
@@ -39,7 +39,7 @@ __wt_block_truncate(WT_SESSION_IMPL *session, WT_BLOCK *block, wt_off_t len)
      * backups, which only copies log files, or targeted backups, stops all block truncation
      * unnecessarily). We may want a more targeted solution at some point.
      */
-    if (!conn->hot_backup) {
+    if (conn->hot_backup_start == 0) {
         WT_WITH_HOTBACKUP_READ_LOCK(session, ret = __wt_ftruncate(session, block->fh, len), NULL);
     }
 
diff --git a/src/third_party/wiredtiger/src/conn/conn_log.c b/src/third_party/wiredtiger/src/conn/conn_log.c
index 347b68d00f4..2120f12e169 100644
--- a/src/third_party/wiredtiger/src/conn/conn_log.c
+++ b/src/third_party/wiredtiger/src/conn/conn_log.c
@@ -567,7 +567,7 @@ __log_file_server(void *arg)
                  * file system may not support truncate: both are OK, it's just more work during
                  * cursor traversal.
                  */
-                if (!conn->hot_backup && conn->log_cursors == 0) {
+                if (conn->hot_backup_start == 0 && conn->log_cursors == 0) {
                     WT_WITH_HOTBACKUP_READ_LOCK(session,
                       WT_ERR_ERROR_OK(__wt_ftruncate(session, close_fh, close_end_lsn.l.offset),
                                                   ENOTSUP),
diff --git a/src/third_party/wiredtiger/src/conn/conn_open.c b/src/third_party/wiredtiger/src/conn/conn_open.c
index d0dff5995b1..2f6632ddf8a 100644
--- a/src/third_party/wiredtiger/src/conn/conn_open.c
+++ b/src/third_party/wiredtiger/src/conn/conn_open.c
@@ -39,6 +39,8 @@ __wt_connection_open(WT_CONNECTION_IMPL *conn, const char *cfg[])
      */
     conn->default_session = session;
 
+    __wt_seconds(session, &conn->ckpt_most_recent);
+
     /*
      * Publish: there must be a barrier to ensure the connection structure fields are set before
      * other threads read from the pointer.
diff --git a/src/third_party/wiredtiger/src/cursor/cur_backup.c b/src/third_party/wiredtiger/src/cursor/cur_backup.c
index a16b7183a9d..1340dfafb95 100644
--- a/src/third_party/wiredtiger/src/cursor/cur_backup.c
+++ b/src/third_party/wiredtiger/src/cursor/cur_backup.c
@@ -311,7 +311,7 @@ __backup_add_id(WT_SESSION_IMPL *session, WT_CONFIG_ITEM *cval)
     for (i = 0; i < WT_BLKINCR_MAX; ++i) {
         blk = &conn->incr_backups[i];
         __wt_verbose(session, WT_VERB_BACKUP, "blk[%u] flags 0x%" PRIx64, i, blk->flags);
-        /* If it isn't use, we can use it. */
+        /* If it isn't already in use, we can use it. */
         if (!F_ISSET(blk, WT_BLKINCR_INUSE))
             break;
     }
@@ -614,7 +614,7 @@ __backup_start(
      * Single thread hot backups: we're holding the schema lock, so we know we'll serialize with
      * other attempts to start a hot backup.
      */
-    if (conn->hot_backup && !is_dup)
+    if (conn->hot_backup_start != 0 && !is_dup)
         WT_RET_MSG(session, EINVAL, "there is already a backup cursor open");
 
     if (F_ISSET(session, WT_SESSION_BACKUP_DUP) && is_dup)
@@ -775,7 +775,7 @@ __backup_stop(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb)
     WT_TRET(__wt_backup_file_remove(session));
 
     /* Checkpoint deletion and next hot backup can proceed. */
-    WT_WITH_HOTBACKUP_WRITE_LOCK(session, conn->hot_backup = false);
+    WT_WITH_HOTBACKUP_WRITE_LOCK(session, conn->hot_backup_start = 0);
     F_CLR(session, WT_SESSION_BACKUP_CURSOR);
 
     return (ret);
diff --git a/src/third_party/wiredtiger/src/docs/backup.dox b/src/third_party/wiredtiger/src/docs/backup.dox
index 610033d05cf..ac18263eff0 100644
--- a/src/third_party/wiredtiger/src/docs/backup.dox
+++ b/src/third_party/wiredtiger/src/docs/backup.dox
@@ -56,10 +56,9 @@ aggregate the file names from the cursor and then list the file names as
 arguments to a file archiver such as the system tar utility.
 
 During the period the backup cursor is open, database checkpoints can
-be created, but no checkpoints can be deleted.  This may result in
-significant file growth. Additionally while the backup cursor is open
-automatic log file archiving, even if enabled, will not reclaim any
-log files.
+be created, but checkpoints created prior to the backup cursor cannot
+be deleted. Additionally while the backup cursor is open automatic log
+file archiving, even if enabled, will not reclaim any log files.
 
 Additionally, if a crash occurs during the period the backup cursor is
 open and logging is disabled (in other words, when depending on
diff --git a/src/third_party/wiredtiger/src/include/connection.h b/src/third_party/wiredtiger/src/include/connection.h
index 1bed17bad88..8106359f934 100644
--- a/src/third_party/wiredtiger/src/include/connection.h
+++ b/src/third_party/wiredtiger/src/include/connection.h
@@ -151,10 +151,10 @@ struct __wt_named_extractor {
  *	Macro to set connection data appropriately for when we commence hot
  *	backup.
  */
-#define WT_CONN_HOTBACKUP_START(conn)   \
-    do {                                \
-        (conn)->hot_backup = true;      \
-        (conn)->hot_backup_list = NULL; \
+#define WT_CONN_HOTBACKUP_START(conn)                        \
+    do {                                                     \
+        (conn)->hot_backup_start = (conn)->ckpt_most_recent; \
+        (conn)->hot_backup_list = NULL;                      \
     } while (0)
 
 /*
@@ -270,13 +270,14 @@ struct __wt_connection_impl {
     WT_TXN_GLOBAL txn_global; /* Global transaction state */
 
     WT_RWLOCK hot_backup_lock; /* Hot backup serialization */
-    bool hot_backup;           /* Hot backup in progress */
+    uint64_t hot_backup_start; /* Clock value of most recent checkpoint needed by hot backup */
     char **hot_backup_list;    /* Hot backup file list */
 
     WT_SESSION_IMPL *ckpt_session; /* Checkpoint thread session */
     wt_thread_t ckpt_tid;          /* Checkpoint thread */
     bool ckpt_tid_set;             /* Checkpoint thread set */
     WT_CONDVAR *ckpt_cond;         /* Checkpoint wait mutex */
+    uint64_t ckpt_most_recent;     /* Clock value of most recent checkpoint */
 #define WT_CKPT_LOGSIZE(conn) ((conn)->ckpt_logsize != 0)
     wt_off_t ckpt_logsize; /* Checkpoint log size period */
     bool ckpt_signalled;   /* Checkpoint signalled */
diff --git a/src/third_party/wiredtiger/src/include/meta.h b/src/third_party/wiredtiger/src/include/meta.h
index 2a27e083b18..594131241ad 100644
--- a/src/third_party/wiredtiger/src/include/meta.h
+++ b/src/third_party/wiredtiger/src/include/meta.h
@@ -77,7 +77,7 @@ struct __wt_blkincr {
 /*
  * At the default granularity, this is enough for blocks in a 2G file.
  */
-#define WT_BLOCK_MODS_LIST_MIN 16 /* Initial bytes for bitmap. */
+#define WT_BLOCK_MODS_LIST_MIN 128 /* Initial bits for bitmap. */
 struct __wt_block_mods {
     const char *id_str;
 
diff --git a/src/third_party/wiredtiger/src/include/schema.h b/src/third_party/wiredtiger/src/include/schema.h
index 69d948e70fd..83827896f43 100644
--- a/src/third_party/wiredtiger/src/include/schema.h
+++ b/src/third_party/wiredtiger/src/include/schema.h
@@ -264,7 +264,7 @@ struct __wt_table {
         if ((skipp) != (bool *)NULL)                            \
             *(bool *)(skipp) = true;                            \
         if (F_ISSET(session, WT_SESSION_LOCKED_HOTBACKUP)) {    \
-            if (!__conn->hot_backup) {                          \
+            if (__conn->hot_backup_start == 0) {                \
                 if ((skipp) != (bool *)NULL)                    \
                     *(bool *)(skipp) = false;                   \
                 op;                                             \
@@ -272,7 +272,7 @@ struct __wt_table {
         } else {                                                \
             __wt_readlock(session, &__conn->hot_backup_lock);   \
             F_SET(session, WT_SESSION_LOCKED_HOTBACKUP_READ);   \
-            if (!__conn->hot_backup) {                          \
+            if (__conn->hot_backup_start == 0) {                \
                 if ((skipp) != (bool *)NULL)                    \
                     *(bool *)(skipp) = false;                   \
                 op;                                             \
diff --git a/src/third_party/wiredtiger/src/include/wiredtiger.in b/src/third_party/wiredtiger/src/include/wiredtiger.in
index d0fb48c1c9a..d9f691e8630 100644
--- a/src/third_party/wiredtiger/src/include/wiredtiger.in
+++ b/src/third_party/wiredtiger/src/include/wiredtiger.in
@@ -1938,8 +1938,9 @@ struct __wt_session {
 	 * one of the following keys: \c "from=all" to drop all checkpoints\, \c "from=<checkpoint>"
 	 * to drop all checkpoints after and including the named checkpoint\, or \c
 	 * "to=<checkpoint>" to drop all checkpoints before and including the named checkpoint.
-	 * Checkpoints cannot be dropped while a hot backup is in progress or if open in a cursor.,
-	 * a list of strings; default empty.}
+	 * Checkpoints cannot be dropped if open in a cursor.  While a hot backup is in progress\,
+	 * checkpoints created prior to the start of the backup cannot be dropped., a list of
+	 * strings; default empty.}
 	 * @config{force, if false (the default)\, checkpoints may be skipped if the underlying
 	 * object has not been modified\, if true\, this option forces the checkpoint., a boolean
 	 * flag; default \c false.}
diff --git a/src/third_party/wiredtiger/src/log/log.c b/src/third_party/wiredtiger/src/log/log.c
index 4c1f6ada07d..9856a18c853 100644
--- a/src/third_party/wiredtiger/src/log/log.c
+++ b/src/third_party/wiredtiger/src/log/log.c
@@ -1166,7 +1166,7 @@ __log_newfile(WT_SESSION_IMPL *session, bool conn_open, bool *created)
      * can copy the files in any way they choose, and a log file rename might confuse things.
      */
     create_log = true;
-    if (conn->log_prealloc > 0 && !conn->hot_backup) {
+    if (conn->log_prealloc > 0 && conn->hot_backup_start == 0) {
         WT_WITH_HOTBACKUP_READ_LOCK(
           session, ret = __log_alloc_prealloc(session, log->fileid), &skipp);
 
@@ -1196,7 +1196,7 @@ __log_newfile(WT_SESSION_IMPL *session, bool conn_open, bool *created)
          * not using pre-allocated log files during backup
          * (see comment above).
          */
-        if (!conn->hot_backup)
+        if (conn->hot_backup_start == 0)
             log->prep_missed++;
         WT_RET(__wt_log_allocfile(session, log->fileid, WT_LOG_FILENAME));
     }
@@ -1385,7 +1385,7 @@ __log_truncate_file(WT_SESSION_IMPL *session, WT_FH *log_fh, wt_off_t offset)
     conn = S2C(session);
     log = conn->log;
 
-    if (!F_ISSET(log, WT_LOG_TRUNCATE_NOTSUP) && !conn->hot_backup) {
+    if (!F_ISSET(log, WT_LOG_TRUNCATE_NOTSUP) && conn->hot_backup_start == 0) {
         WT_WITH_HOTBACKUP_READ_LOCK(session, ret = __wt_ftruncate(session, log_fh, offset), &skipp);
         if (!skipp) {
             if (ret != ENOTSUP)
diff --git a/src/third_party/wiredtiger/src/meta/meta_ckpt.c b/src/third_party/wiredtiger/src/meta/meta_ckpt.c
index 341c1b8b5f7..4d23e155363 100644
--- a/src/third_party/wiredtiger/src/meta/meta_ckpt.c
+++ b/src/third_party/wiredtiger/src/meta/meta_ckpt.c
@@ -462,9 +462,10 @@ __wt_meta_ckptlist_get(
     WT_CKPT *ckpt, *ckptbase;
     WT_CONFIG ckptconf;
     WT_CONFIG_ITEM k, v;
+    WT_CONNECTION_IMPL *conn;
     WT_DECL_RET;
     size_t allocated, slot;
-    int64_t maxorder;
+    uint64_t most_recent;
     char *config;
 
     *ckptbasep = NULL;
@@ -472,6 +473,7 @@ __wt_meta_ckptlist_get(
     ckptbase = NULL;
     allocated = slot = 0;
     config = NULL;
+    conn = S2C(session);
 
     /* Retrieve the metadata information for the file. */
     WT_RET(__wt_metadata_search(session, fname, &config));
@@ -509,13 +511,21 @@ __wt_meta_ckptlist_get(
         WT_ERR(__wt_realloc_def(session, &allocated, slot + 2, &ckptbase));
 
         /* The caller may be adding a value, initialize it. */
-        maxorder = 0;
-        WT_CKPT_FOREACH (ckptbase, ckpt)
-            if (ckpt->order > maxorder)
-                maxorder = ckpt->order;
-        ckpt->order = maxorder + 1;
+        ckpt = &ckptbase[slot];
+        ckpt->order = (slot == 0) ? 1 : ckptbase[slot - 1].order + 1;
         __wt_seconds(session, &ckpt->sec);
         /*
+         * Update time value for most recent checkpoint, not letting it move backwards. It is
+         * possible to race here, so use atomic CAS. This code relies on the fact that anyone we
+         * race with will only increase (never decrease) the most recent checkpoint time value.
+         */
+        for (;;) {
+            WT_ORDERED_READ(most_recent, conn->ckpt_most_recent);
+            if (ckpt->sec <= most_recent ||
+              __wt_atomic_cas64(&conn->ckpt_most_recent, most_recent, ckpt->sec))
+                break;
+        }
+        /*
          * Load most recent checkpoint backup blocks to this checkpoint.
          */
         WT_ERR(__ckpt_load_blk_mods(session, config, ckpt));
@@ -527,7 +537,7 @@ __wt_meta_ckptlist_get(
          * the checkpoint's modified blocks from the block manager.
          */
         F_SET(ckpt, WT_CKPT_ADD);
-        if (F_ISSET(S2C(session), WT_CONN_INCR_BACKUP)) {
+        if (F_ISSET(conn, WT_CONN_INCR_BACKUP)) {
             F_SET(ckpt, WT_CKPT_BLOCK_MODS);
             WT_ERR(__ckpt_valid_blk_mods(session, ckpt));
         }
diff --git a/src/third_party/wiredtiger/src/schema/schema_util.c b/src/third_party/wiredtiger/src/schema/schema_util.c
index 25ef013648c..86bff7b086c 100644
--- a/src/third_party/wiredtiger/src/schema/schema_util.c
+++ b/src/third_party/wiredtiger/src/schema/schema_util.c
@@ -26,7 +26,7 @@ __schema_backup_check_int(WT_SESSION_IMPL *session, const char *name)
      * There is a window at the end of a backup where the list has been cleared from the connection
      * but the flag is still set. It is safe to drop at that point.
      */
-    if (!conn->hot_backup || (backup_list = conn->hot_backup_list) == NULL) {
+    if (conn->hot_backup_start == 0 || (backup_list = conn->hot_backup_list) == NULL) {
         return (0);
     }
     for (i = 0; backup_list[i] != NULL; ++i) {
@@ -50,7 +50,7 @@ __wt_schema_backup_check(WT_SESSION_IMPL *session, const char *name)
     WT_DECL_RET;
 
     conn = S2C(session);
-    if (!conn->hot_backup)
+    if (conn->hot_backup_start == 0)
         return (0);
     WT_WITH_HOTBACKUP_READ_LOCK_UNCOND(session, ret = __schema_backup_check_int(session, name));
     return (ret);
diff --git a/src/third_party/wiredtiger/src/txn/txn_ckpt.c b/src/third_party/wiredtiger/src/txn/txn_ckpt.c
index 5c61faee070..cba8357db23 100644
--- a/src/third_party/wiredtiger/src/txn/txn_ckpt.c
+++ b/src/third_party/wiredtiger/src/txn/txn_ckpt.c
@@ -1112,7 +1112,6 @@ static void
 __drop(WT_CKPT *ckptbase, const char *name, size_t len)
 {
     WT_CKPT *ckpt;
-    u_int max_ckpt_drop;
 
     /*
      * If we're dropping internal checkpoints, match to the '.' separating the checkpoint name from
@@ -1121,20 +1120,9 @@ __drop(WT_CKPT *ckptbase, const char *name, size_t len)
      * it's one we want to drop.
      */
     if (strncmp(WT_CHECKPOINT, name, len) == 0) {
-        /*
-         * Currently, hot backup cursors block checkpoint drop, which means releasing a hot backup
-         * cursor can result in immediately attempting to drop lots of checkpoints, which involves a
-         * fair amount of work while holding locks. Limit the number of standard checkpoints dropped
-         * per checkpoint.
-         */
-        max_ckpt_drop = 0;
         WT_CKPT_FOREACH (ckptbase, ckpt)
-            if (WT_PREFIX_MATCH(ckpt->name, WT_CHECKPOINT)) {
+            if (WT_PREFIX_MATCH(ckpt->name, WT_CHECKPOINT))
                 F_SET(ckpt, WT_CKPT_DELETE);
-#define WT_MAX_CHECKPOINT_DROP 4
-                if (++max_ckpt_drop >= WT_MAX_CHECKPOINT_DROP)
-                    break;
-            }
     } else
         WT_CKPT_FOREACH (ckptbase, ckpt)
             if (WT_STRING_MATCH(ckpt->name, name, len))
@@ -1214,30 +1202,44 @@ __checkpoint_lock_dirty_tree_int(WT_SESSION_IMPL *session, bool is_checkpoint, b
 {
     WT_CONNECTION_IMPL *conn;
     WT_DECL_RET;
+    u_int max_ckpt_drop;
+    bool is_wt_ckpt;
 
     WT_UNUSED(is_checkpoint);
     conn = S2C(session);
 
-    /*
-     * We can't delete checkpoints if a backup cursor is open. WiredTiger checkpoints are uniquely
-     * named and it's OK to have multiple of them in the system: clear the delete flag for them, and
-     * otherwise fail. Hold the lock until we're done (blocking hot backups from starting), we don't
-     * want to race with a future hot backup.
-     */
-    if (conn->hot_backup)
-        WT_CKPT_FOREACH (ckptbase, ckpt) {
-            if (!F_ISSET(ckpt, WT_CKPT_DELETE))
-                continue;
-            if (WT_PREFIX_MATCH(ckpt->name, WT_CHECKPOINT)) {
+    /* Check that it is OK to remove all the checkpoints marked for deletion. */
+    max_ckpt_drop = 0;
+    WT_CKPT_FOREACH (ckptbase, ckpt) {
+        if (!F_ISSET(ckpt, WT_CKPT_DELETE))
+            continue;
+        is_wt_ckpt = WT_PREFIX_MATCH(ckpt->name, WT_CHECKPOINT);
+
+        /*
+         * If there is a hot backup, don't delete any WiredTiger checkpoint that could possibly have
+         * been created before the backup started. Fail if trying to delete any other named
+         * checkpoint.
+         */
+        if (conn->hot_backup_start != 0 && ckpt->sec <= conn->hot_backup_start) {
+            if (is_wt_ckpt) {
                 F_CLR(ckpt, WT_CKPT_DELETE);
                 continue;
             }
             WT_RET_MSG(session, EBUSY,
-              "checkpoint %s blocked by hot backup: it would"
-              "delete an existing checkpoint, and checkpoints "
-              "cannot be deleted during a hot backup",
+              "checkpoint %s blocked by hot backup: it would "
+              "delete an existing named checkpoint, and such "
+              "checkpoints cannot be deleted during a hot backup",
               ckpt->name);
         }
+        /*
+         * Dropping checkpoints involves a fair amount of work while holding locks. Limit the number
+         * of WiredTiger checkpoints dropped per checkpoint.
+         */
+        if (is_wt_ckpt)
+#define WT_MAX_CHECKPOINT_DROP 4
+            if (++max_ckpt_drop >= WT_MAX_CHECKPOINT_DROP)
+                F_CLR(ckpt, WT_CKPT_DELETE);
+    }
 
     /*
      * Mark old checkpoints that are being deleted and figure out which trees we can skip in this
@@ -1257,6 +1259,8 @@ __checkpoint_lock_dirty_tree_int(WT_SESSION_IMPL *session, bool is_checkpoint, b
         WT_CKPT_FOREACH (ckptbase, ckpt) {
             if (!F_ISSET(ckpt, WT_CKPT_DELETE))
                 continue;
+            WT_ASSERT(session, !WT_PREFIX_MATCH(ckpt->name, WT_CHECKPOINT) ||
+                conn->hot_backup_start == 0 || ckpt->sec > conn->hot_backup_start);
             /*
              * We can't delete checkpoints referenced by a cursor. WiredTiger checkpoints are
              * uniquely named and it's OK to have multiple in the system: clear the delete flag for
diff --git a/src/third_party/wiredtiger/src/txn/txn_log.c b/src/third_party/wiredtiger/src/txn/txn_log.c
index 195c4e336e5..dae4a2ba166 100644
--- a/src/third_party/wiredtiger/src/txn/txn_log.c
+++ b/src/third_party/wiredtiger/src/txn/txn_log.c
@@ -550,8 +550,9 @@ __wt_txn_checkpoint_log(WT_SESSION_IMPL *session, bool full, uint32_t flags, WT_
          * connection close, only during a full checkpoint. A clean close may not update any
          * metadata LSN and we do not want to archive in that case.
          */
-        if (!conn->hot_backup && (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_RECOVER_DIRTY) ||
-                                   FLD_ISSET(conn->log_flags, WT_CONN_LOG_FORCE_DOWNGRADE)) &&
+        if (conn->hot_backup_start == 0 &&
+          (!FLD_ISSET(conn->log_flags, WT_CONN_LOG_RECOVER_DIRTY) ||
+              FLD_ISSET(conn->log_flags, WT_CONN_LOG_FORCE_DOWNGRADE)) &&
           txn->full_ckpt)
             __wt_log_ckpt(session, ckpt_lsn);
 
diff --git a/src/third_party/wiredtiger/test/format/config.c b/src/third_party/wiredtiger/test/format/config.c
index 54a09229ce4..eb86c0860e7 100644
--- a/src/third_party/wiredtiger/test/format/config.c
+++ b/src/third_party/wiredtiger/test/format/config.c
@@ -264,8 +264,13 @@ static void
 config_backup_incr(void)
 {
     /* Incremental backup requires backup. */
-    if (g.c_backups == 0)
+    if (g.c_backups == 0) {
+        if (!config_is_perm("backup.incremental"))
+            config_single("backup.incremental=off", false);
+        if (g.c_backup_incr_flag != INCREMENTAL_OFF)
+            testutil_die(EINVAL, "backup.incremental requires backups be configured");
         return;
+    }
 
     /*
      * Incremental backup using log files is incompatible with logging archival. Testing log file
@@ -298,6 +303,7 @@ config_backup_incr(void)
             if (g.c_logging_archive)
                 config_single("logging.archive=0", false);
             config_single("backup.incremental=log", false);
+            break;
         }
     /* FALLTHROUGH */
     case 7: /* 40% block based incremental */
diff --git a/src/third_party/wiredtiger/test/suite/test_backup01.py b/src/third_party/wiredtiger/test/suite/test_backup01.py
index ff4dd439189..00a3ccaf0f6 100644
--- a/src/third_party/wiredtiger/test/suite/test_backup01.py
+++ b/src/third_party/wiredtiger/test/suite/test_backup01.py
@@ -30,6 +30,7 @@ import glob
 import os
 import shutil
 import string
+import time
 from suite_subprocess import suite_subprocess
 import wiredtiger, wttest
 from wtdataset import SimpleDataSet, ComplexDataSet, ComplexLSMDataSet
@@ -163,8 +164,7 @@ class test_backup(wttest.WiredTigerTestCase, suite_subprocess):
         self.assertEqual(ret, wiredtiger.WT_NOTFOUND)
         self.assertEqual(i, total)
 
-    # Test that named checkpoints can't be deleted while backup cursors are
-    # open, but that normal checkpoints continue to work.
+    # Test interaction between checkpoints and a backup cursor.
     def test_checkpoint_delete(self):
         # You cannot name checkpoints including LSM tables, skip those.
         self.populate(1)
@@ -177,7 +177,8 @@ class test_backup(wttest.WiredTigerTestCase, suite_subprocess):
             self.objs[0][0], None, "checkpoint=one"))
 
         # Confirm opening a backup cursor causes checkpoint to fail if dropping
-        # a named checkpoint, but does not stop a default checkpoint.
+        # a named checkpoint created before the backup cursor, but does not stop a
+        # default checkpoint.
         cursor = self.session.open_cursor('backup:', None, None)
         self.session.checkpoint()
         msg = '/checkpoints cannot be deleted during a hot backup/'
@@ -187,7 +188,23 @@ class test_backup(wttest.WiredTigerTestCase, suite_subprocess):
         self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
             lambda: self.session.checkpoint("name=three,drop=(two)"), msg)
         self.session.checkpoint()
+
+        # Need to pause a couple seconds; checkpoints that are assigned the same timestamp as
+        # the backup will be pinned, even if they occur after the backup starts.
+        time.sleep(2)
+        self.session.checkpoint("name=four")
+        self.session.checkpoint("drop=(four)")
+        self.assertRaises(wiredtiger.WiredTigerError,
+            lambda: self.session.open_cursor(
+            self.objs[0][0], None, "checkpoint=four"))
+
+        # Confirm that after closing the backup cursor the original named checkpoint can
+        # be deleted.
         cursor.close()
+        self.session.checkpoint("drop=(two)")
+        self.assertRaises(wiredtiger.WiredTigerError,
+            lambda: self.session.open_cursor(
+            self.objs[0][0], None, "checkpoint=two"))
 
 if __name__ == '__main__':
     wttest.run()
diff --git a/src/third_party/wiredtiger/test/suite/test_backup14.py b/src/third_party/wiredtiger/test/suite/test_backup14.py
index 7a2ec4f427f..c312020bcef 100644
--- a/src/third_party/wiredtiger/test/suite/test_backup14.py
+++ b/src/third_party/wiredtiger/test/suite/test_backup14.py
@@ -175,9 +175,9 @@ class test_backup14(wttest.WiredTigerTestCase, suite_subprocess):
                     shutil.copy(copy_from, copy_to)
                 else:
                     self.pr('Range copy file ' + newfile + ' offset ' + str(offset) + ' len ' + str(size))
-                    write_from = newfile
+                    read_from = newfile
                     write_to = self.home_incr + '.' + str(self.counter) + '/' + newfile
-                    rfp = open(write_from, "r+b")
+                    rfp = open(read_from, "r+b")
                     wfp = open(write_to, "w+b")
                     rfp.seek(offset, 0)
                     wfp.seek(offset, 0)
diff --git a/src/third_party/wiredtiger/test/suite/test_backup15.py b/src/third_party/wiredtiger/test/suite/test_backup15.py
new file mode 100644
index 00000000000..08ddf177ea6
--- /dev/null
+++ b/src/third_party/wiredtiger/test/suite/test_backup15.py
@@ -0,0 +1,315 @@
+#!/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.
+
+import wiredtiger, wttest
+import os, shutil
+from helper import compare_files
+from suite_subprocess import suite_subprocess
+from wtdataset import simple_key
+from wtscenario import make_scenarios
+import glob
+
+# test_backup15.py
+# Test cursor backup with a block-based incremental cursor.
+class test_backup15(wttest.WiredTigerTestCase, suite_subprocess):
+    bkp_home = "WT_BLOCK"
+    counter=0
+    conn_config='cache_size=1G,log=(enabled,file_max=100K)'
+    logmax="100K"
+    max_iteration=7
+    mult=0
+    nops=100000
+    savefirst=0
+    savekey='NOTSET'
+    uri="table:main"
+
+    dir='backup.dir'                    # Backup directory name
+    home_full = "WT_BLOCK_LOG_FULL"
+    home_incr = "WT_BLOCK_LOG_INCR"
+
+    full_out = "./backup_block_full"
+    incr_out = "./backup_block_incr"
+    logpath = "logpath"
+    new_table=False
+    initial_backup=False
+
+    pfx = 'test_backup'
+    # Set the key and value big enough that we modify a few blocks.
+    bigkey = 'Key' * 100
+    bigval = 'Value' * 100
+
+    #
+    # Set up all the directories needed for the test. We have a full backup directory for each
+    # iteration and an incremental backup for each iteration. That way we can compare the full and
+    # incremental each time through.
+    #
+    def setup_directories(self):
+        for i in range(0, self.max_iteration):
+            remove_dir = self.home_incr + '.' + str(i)
+
+            create_dir = self.home_incr + '.' + str(i) + '/' + self.logpath
+            if os.path.exists(remove_dir):
+                os.remove(remove_dir)
+            os.makedirs(create_dir)
+
+            if i == 0:
+                continue
+            remove_dir = self.home_full + '.' + str(i)
+            create_dir = self.home_full + '.' + str(i) + '/' + self.logpath
+            if os.path.exists(remove_dir):
+                os.remove(remove_dir)
+            os.makedirs(create_dir)
+
+    def take_full_backup(self):
+        if self.counter != 0:
+            hdir = self.home_full + '.' + str(self.counter)
+        else:
+            hdir = self.home_incr
+
+        #
+        # First time through we take a full backup into the incremental directories. Otherwise only
+        # into the appropriate full directory.
+        #
+        buf = None
+        if self.initial_backup == True:
+            buf = 'incremental=(granularity=1M,enabled=true,this_id=ID0)'
+
+        cursor = self.session.open_cursor('backup:', None, buf)
+        while True:
+            ret = cursor.next()
+            if ret != 0:
+                break
+            newfile = cursor.get_key()
+
+            if self.counter == 0:
+                # Take a full bakcup into each incremental directory
+                for i in range(0, self.max_iteration):
+                    copy_from = newfile
+                    # If it is log file, prepend the path.
+                    if ("WiredTigerLog" in newfile):
+                        copy_to = self.home_incr + '.' + str(i) + '/' + self.logpath
+                    else:
+                        copy_to = self.home_incr + '.' + str(i)
+                    shutil.copy(copy_from, copy_to)
+            else:
+                copy_from = newfile
+                # If it is log file, prepend the path.
+                if ("WiredTigerLog" in newfile):
+                    copy_to = hdir + '/' + self.logpath
+                else:
+                    copy_to = hdir
+
+                shutil.copy(copy_from, copy_to)
+        self.assertEqual(ret, wiredtiger.WT_NOTFOUND)
+        cursor.close()
+
+    def take_incr_backup(self):
+        # Open the backup data source for incremental backup.
+        buf = 'incremental=(src_id="ID' +  str(self.counter-1) + '",this_id="ID' + str(self.counter) + '")'
+        self.pr(buf)
+        bkup_c = self.session.open_cursor('backup:', None, buf)
+        while True:
+            ret = bkup_c.next()
+            if ret != 0:
+                break
+            newfile = bkup_c.get_key()
+            h = self.home_incr + '.0'
+            copy_from = newfile
+            # If it is log file, prepend the path.
+            if ("WiredTigerLog" in newfile):
+                copy_to = h + '/' + self.logpath
+            else:
+                copy_to = h
+
+            shutil.copy(copy_from, copy_to)
+            first = True
+            config = 'incremental=(file=' + newfile + ')'
+            dup_cnt = 0
+            incr_c = self.session.open_cursor(None, bkup_c, config)
+
+            # For each file listed, open a duplicate backup cursor and copy the blocks.
+            while True:
+                ret = incr_c.next()
+                if ret != 0:
+                    break
+                incrlist = incr_c.get_keys()
+                offset = incrlist[0]
+                size = incrlist[1]
+                curtype = incrlist[2]
+                # 1 is WT_BACKUP_FILE
+                # 2 is WT_BACKUP_RANGE
+                self.assertTrue(curtype == 1 or curtype == 2)
+                if curtype == 1:
+                    if first == True:
+                        h = self.home_incr + '.' + str(self.counter)
+                        first = False
+
+                    copy_from = newfile
+                    if ("WiredTigerLog" in newfile):
+                        copy_to = h + '/' + self.logpath
+                    else:
+                        copy_to = h
+                    shutil.copy(copy_from, copy_to)
+                else:
+                    self.pr('Range copy file ' + newfile + ' offset ' + str(offset) + ' len ' + str(size))
+                    read_from = newfile
+                    if self.counter > 0:
+                        old_to = self.home_incr + '.' + str(self.counter - 1) + '/' + newfile
+                    else:
+                        old_to = newfile
+                    write_to = self.home_incr + '.' + str(self.counter) + '/' + newfile
+                    write_to = self.home_incr + '.' + str(self.counter) + '/' + newfile
+                    rfp = open(read_from, "r+b")
+                    self.pr('RANGE CHECK file ' + old_to + ' offset ' + str(offset) + ' len ' + str(size))
+                    rfp2 = open(old_to, "r+b")
+                    rfp.seek(offset, 0)
+                    rfp2.seek(offset, 0)
+                    buf = rfp.read(size)
+                    buf2 = rfp2.read(size)
+                    self.assertNotEqual(buf, buf2)
+                    wfp = open(write_to, "w+b")
+                    wfp.seek(offset, 0)
+                    wfp.write(buf)
+                    rfp.close()
+                    rfp2.close()
+                    wfp.close()
+                dup_cnt += 1
+            self.assertEqual(ret, wiredtiger.WT_NOTFOUND)
+            incr_c.close()
+
+            # For each file, we want to copy the file into each of the later incremental directories
+            for i in range(self.counter, self.max_iteration):
+                h = self.home_incr + '.' + str(i)
+                copy_from = newfile
+                if ("WiredTigerLog" in newfile):
+                    copy_to = h + '/' + self.logpath
+                else:
+                    copy_to = h
+                shutil.copy(copy_from, copy_to)
+        self.assertEqual(ret, wiredtiger.WT_NOTFOUND)
+        bkup_c.close()
+
+    def compare_backups(self, t_uri):
+        #
+        # Run wt dump on full backup directory
+        #
+        full_backup_out = self.full_out + '.' + str(self.counter)
+        home_dir = self.home_full + '.' + str(self.counter)
+        if self.counter == 0:
+            home_dir = self.home
+
+        self.runWt(['-R', '-h', home_dir, 'dump', t_uri], outfilename=full_backup_out)
+        #
+        # Run wt dump on incremental backup directory
+        #
+        incr_backup_out = self.incr_out + '.' + str(self.counter)
+        home_dir = self.home_incr + '.' + str(self.counter)
+        self.runWt(['-R', '-h', home_dir, 'dump', t_uri], outfilename=incr_backup_out)
+
+        self.assertEqual(True,
+            compare_files(self, full_backup_out, incr_backup_out))
+
+    #
+    # Add data to the given uri.
+    #
+    def add_data(self, uri):
+        c = self.session.open_cursor(uri, None, None)
+        # The first time we want to add in a lot of data. Then after that we want to
+        # rapidly change a single key to create a hotspot in one block.
+        if self.savefirst < 2:
+            nops = self.nops
+        else:
+            nops = self.nops // 10
+        for i in range(0, nops):
+            num = i + (self.mult * nops)
+            if self.savefirst >= 2:
+                key = self.savekey
+            else:
+                key = str(num) + self.bigkey + str(num)
+            val = str(num) + self.bigval + str(num)
+            c[key] = val
+        if self.savefirst == 0:
+            self.savekey = key
+        self.savefirst += 1
+        c.close()
+
+        # Increase the multiplier so that later calls insert unique items.
+        self.mult += 1
+        # Increase the counter so that later backups have unique ids.
+        if self.initial_backup == False:
+            self.counter += 1
+
+    #
+    # This function will add records to the table (table:main), take incremental/full backups and
+    # validate the backups.
+    #
+    def add_data_validate_backups(self):
+        self.pr('Adding initial data')
+        self.initial_backup = True
+        self.add_data(self.uri)
+        self.take_full_backup()
+        self.initial_backup = False
+        self.session.checkpoint()
+
+        # Each call now to take a full backup will make a copy into a full directory. Then
+        # each incremental will take an incremental backup and we can compare them.
+        self.add_data(self.uri)
+        self.session.checkpoint()
+        self.take_full_backup()
+        self.take_incr_backup()
+        self.compare_backups(self.uri)
+
+        self.add_data(self.uri)
+        self.session.checkpoint()
+        self.take_incr_backup()
+        self.take_full_backup()
+        self.compare_backups(self.uri)
+
+        self.add_data(self.uri)
+        self.session.checkpoint()
+        self.take_full_backup()
+        self.take_incr_backup()
+        self.compare_backups(self.uri)
+
+        self.add_data(self.uri)
+        self.take_incr_backup()
+        self.take_full_backup()
+        self.compare_backups(self.uri)
+
+    def test_backup15(self):
+        os.mkdir(self.bkp_home)
+        self.home = self.bkp_home
+        self.session.create(self.uri, "key_format=S,value_format=S")
+
+        self.setup_directories()
+
+        self.pr('*** Add data, checkpoint, take backups and validate ***')
+        self.add_data_validate_backups()
+
+if __name__ == '__main__':
+    wttest.run()
diff --git a/src/third_party/wiredtiger/test/suite/test_checkpoint05.py b/src/third_party/wiredtiger/test/suite/test_checkpoint05.py
new file mode 100644
index 00000000000..58af3003a60
--- /dev/null
+++ b/src/third_party/wiredtiger/test/suite/test_checkpoint05.py
@@ -0,0 +1,84 @@
+#!/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.
+#
+# test_checkpoint05.py
+# Verify that we don't accumulate a lot of checkpoints while a backup
+# cursor is open. WiredTiger checkpoints created after the backup cursor
+# should get deleted as usual.
+
+import time
+import wiredtiger, wttest
+
+class test_checkpoint05(wttest.WiredTigerTestCase):
+    conn_config = 'create,cache_size=100MB,log=(archive=false,enabled=true,file_max=100K)'
+
+    def count_checkpoints(self):
+        metadata_cursor = self.session.open_cursor('metadata:', None, None)
+
+        nckpt = 0
+        while metadata_cursor.next() == 0:
+            key = metadata_cursor.get_key()
+            value = metadata_cursor[key]
+            nckpt = nckpt + value.count("WiredTigerCheckpoint")
+        metadata_cursor.close()
+        return nckpt
+
+    def test_checkpoints_during_backup(self):
+        self.uri = 'table:ckpt05'
+        self.session.create(self.uri, 'key_format=i,value_format=i')
+
+        # Setup: Insert some data and checkpoint it
+        cursor = self.session.open_cursor(self.uri, None)
+        for i in range(16):
+            cursor[i] = i
+        self.session.checkpoint(None)
+
+        # Create backup and check how many checkpoints we have.
+        backup_cursor = self.session.open_cursor('backup:', None, None)
+        initial_count = self.count_checkpoints()
+
+        # Checkpoints created immediately after a backup cursor may get pinned.
+        # Pause to avoid this.
+        time.sleep(2)
+
+        # Take a bunch of checkpoints.
+        for i in range (50):
+            self.session.checkpoint('force=true')
+        cursor.close()
+
+        # There may be a few more checkpoints than when we opened the
+        # backup cursor, but not too many more.  The factor of three
+        # is generous.  But if WT isn't deleting checkpoints there would
+        # be about 30x more checkpoints here.
+        final_count = self.count_checkpoints()
+        self.assertTrue (final_count < initial_count * 3)
+
+        self.session.close()
+
+if __name__ == '__main__':
+    wttest.run()