Merge pull request #558 from wiredtiger/hot-backup-schema

Update backup to use __wt_schema_worker. Stop LSM drops during backups.

14 files changed, 413 insertions, 300 deletions

diff --git a/src/conn/conn_api.c b/src/conn/conn_api.c
index aad877a9d29..0046faf25fb 100644
--- a/src/conn/conn_api.c
+++ b/src/conn/conn_api.c
@@ -862,7 +862,6 @@ wiredtiger_open(const char *home, WT_EVENT_HANDLER *event_handler,
 	WT_ITEM *cbuf, expath, exconfig;
 	WT_SESSION_IMPL *session;
 	const char *cfg[5];
-	int exist;
 
 	*wt_connp = NULL;
 	session = NULL;
@@ -1011,22 +1010,9 @@ wiredtiger_open(const char *home, WT_EVENT_HANDLER *event_handler,
 	 * (which avoids application threads racing to create the metadata file
 	 * later).
 	 */
-	WT_ERR(__wt_meta_turtle_init(session, &exist));
-	if (!exist) {
-		/*
-		 * We're single-threaded, but acquire the schema lock
-		 * regardless: the lower level code checks that it is
-		 * appropriately synchronized.
-		 */
-		WT_WITH_SCHEMA_LOCK(session,
-		    ret = __wt_schema_create(session, WT_METADATA_URI, NULL));
-		WT_ERR(ret);
-	}
+	WT_ERR(__wt_turtle_init(session));
 	WT_ERR(__wt_metadata_open(session));
 
-	/* If there's a hot-backup file, load it. */
-	WT_ERR(__wt_metadata_load_backup(session));
-
 	STATIC_ASSERT(offsetof(WT_CONNECTION_IMPL, iface) == 0);
 	*wt_connp = &conn->iface;
 
diff --git a/src/cursor/cur_backup.c b/src/cursor/cur_backup.c
index 6e123c0589e..53b1729aa0f 100644
--- a/src/cursor/cur_backup.c
+++ b/src/cursor/cur_backup.c
@@ -7,20 +7,16 @@
 
 #include "wt_internal.h"
 
-static int __backup_all(WT_SESSION_IMPL *, WT_CURSOR_BACKUP *, FILE *);
-static int __backup_file_create(WT_SESSION_IMPL *, FILE **);
+static int __backup_all(WT_SESSION_IMPL *, WT_CURSOR_BACKUP *);
+static int __backup_file_create(WT_SESSION_IMPL *, WT_CURSOR_BACKUP *);
 static int __backup_file_remove(WT_SESSION_IMPL *);
 static int __backup_list_append(
     WT_SESSION_IMPL *, WT_CURSOR_BACKUP *, const char *);
 static int __backup_start(
     WT_SESSION_IMPL *, WT_CURSOR_BACKUP *, const char *[]);
 static int __backup_stop(WT_SESSION_IMPL *);
-static int __backup_table(
-    WT_SESSION_IMPL *, WT_CURSOR_BACKUP *, const char *, FILE *);
-static int __backup_table_element(WT_SESSION_IMPL *,
-    WT_CURSOR_BACKUP *, WT_CURSOR *, const char *, const char *, FILE *);
 static int __backup_uri(
-    WT_SESSION_IMPL *, WT_CURSOR_BACKUP *, const char *[], FILE *, int *);
+    WT_SESSION_IMPL *, WT_CURSOR_BACKUP *, const char *[], int *);
 
 /*
  * __curbackup_next --
@@ -96,6 +92,7 @@ __curbackup_close(WT_CURSOR *cursor)
 	}
 
 	ret = __wt_cursor_close(cursor);
+	session->bkp_cursor = NULL;
 
 	WT_WITH_SCHEMA_LOCK(session,
 	    tret = __backup_stop(session));		/* Stop the backup. */
@@ -138,6 +135,7 @@ __wt_curbackup_open(WT_SESSION_IMPL *session,
 	cursor = &cb->iface;
 	*cursor = iface;
 	cursor->session = &session->iface;
+	session->bkp_cursor = cb;
 
 	cursor->key_format = "S";	/* Return the file names as the key. */
 	cursor->value_format = "";	/* No value. */
@@ -168,14 +166,12 @@ static int
 __backup_start(
     WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb, const char *cfg[])
 {
-	FILE *bfp;
 	WT_CONNECTION_IMPL *conn;
 	WT_DECL_RET;
 	int target_list;
 
 	conn = S2C(session);
 
-	bfp = NULL;
 	cb->next = 0;
 	cb->list = NULL;
 
@@ -203,24 +199,28 @@ __backup_start(
 	__wt_spin_unlock(session, &conn->hot_backup_lock);
 
 	/* Create the hot backup file. */
-	WT_ERR(__backup_file_create(session, &bfp));
+	WT_ERR(__backup_file_create(session, cb));
 
 	/*
 	 * If a list of targets was specified, work our way through them.
 	 * Else, generate a list of all database objects.
 	 */
 	target_list = 0;
-	WT_ERR(__backup_uri(session, cb, cfg, bfp, &target_list));
+	WT_ERR(__backup_uri(session, cb, cfg, &target_list));
 	if (!target_list)
-		WT_ERR(__backup_all(session, cb, bfp));
+		WT_ERR(__backup_all(session, cb));
+
+	/* Add the hot backup and single-threading file to the list. */
+	WT_ERR(__backup_list_append(session, cb, WT_METADATA_BACKUP));
+	WT_ERR(__backup_list_append(session, cb, WT_SINGLETHREAD));
 
 	/* Close the hot backup file. */
-	ret = fclose(bfp);
-	bfp = NULL;
+	ret = fclose(cb->bfp);
+	cb->bfp = NULL;
 	WT_ERR_TEST(ret == EOF, __wt_errno());
 
-err:	if (bfp != NULL)
-		WT_TRET(fclose(bfp) == 0 ? 0 : __wt_errno());
+err:	if (cb->bfp != NULL)
+		WT_TRET(fclose(cb->bfp) == 0 ? 0 : __wt_errno());
 
 	if (ret != 0)
 		WT_TRET(__backup_stop(session));
@@ -256,16 +256,15 @@ __backup_stop(WT_SESSION_IMPL *session)
  *	Backup all objects in the database.
  */
 static int
-__backup_all(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb, FILE *bfp)
+__backup_all(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb)
 {
 	WT_CONFIG_ITEM cval;
 	WT_CURSOR *cursor;
 	WT_DECL_RET;
 	int cmp;
-	const char *key, *path, *uri, *value;
+	const char *key, *uri, *value;
 
 	cursor = NULL;
-	path = NULL;
 
 	/*
 	 * Open a cursor on the metadata file and copy all of the entries to
@@ -275,8 +274,8 @@ __backup_all(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb, FILE *bfp)
 	while ((ret = cursor->next(cursor)) == 0) {
 		WT_ERR(cursor->get_key(cursor, &key));
 		WT_ERR(cursor->get_value(cursor, &value));
-		WT_ERR_TEST(
-		    (fprintf(bfp, "%s\n%s\n", key, value) < 0), __wt_errno());
+		WT_ERR_TEST((fprintf(
+		    cb->bfp, "%s\n%s\n", key, value) < 0), __wt_errno());
 
 		/*
 		 * While reading the metadata file, check there are no "sources"
@@ -319,15 +318,8 @@ __backup_all(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb, FILE *bfp)
 	}
 	WT_ERR_NOTFOUND_OK(ret);
 
-	/* Add the hot backup and single-threading file to the list. */
-	WT_ERR(__backup_list_append(session, cb, WT_METADATA_BACKUP));
-	WT_ERR(__backup_list_append(session, cb, WT_SINGLETHREAD));
-
-err:
-	if (cursor != NULL)
+err:	if (cursor != NULL)
 		WT_TRET(cursor->close(cursor));
-	if (path != NULL)
-		__wt_free(session, path);
 	return (ret);
 }
 
@@ -337,19 +329,16 @@ err:
  */
 static int
 __backup_uri(WT_SESSION_IMPL *session,
-    WT_CURSOR_BACKUP *cb, const char *cfg[], FILE *bfp, int *foundp)
+    WT_CURSOR_BACKUP *cb, const char *cfg[], int *foundp)
 {
 	WT_CONFIG targetconf;
 	WT_CONFIG_ITEM cval, k, v;
 	WT_DECL_ITEM(tmp);
 	WT_DECL_RET;
 	int target_list;
-	const char *path, *uri, *value;
+	const char *uri;
 
-	*foundp = 0;
-
-	path = NULL;
-	target_list = 0;
+	*foundp = target_list = 0;
 
 	/*
 	 * If we find a non-empty target configuration string, we have a job,
@@ -372,117 +361,8 @@ __backup_uri(WT_SESSION_IMPL *session,
 			    "%s: invalid backup target: URIs may need quoting",
 			    uri);
 
-		if (WT_PREFIX_MATCH(uri, "file:")) {
-			/* Copy metadata file information to the backup file. */
-			WT_ERR(__wt_metadata_read(session, uri, &value));
-			WT_ERR_TEST((fprintf(bfp,
-				"%s\n%s\n", uri, value) < 0), __wt_errno());
-
-			WT_ERR(__backup_list_append(
-			    session, cb, uri + strlen("file:")));
-			continue;
-		}
-		if (WT_PREFIX_MATCH(uri, "table:")) {
-			WT_ERR(__backup_table(
-			    session, cb, uri + strlen("table:"), bfp));
-			continue;
-		}
-
-		/*
-		 * We only support file and table targets (this is where we'll
-		 * fail if an LSM target is specified).
-		 */
-		WT_ERR_MSG(session,
-		    ENOTSUP, "%s: unsupported backup target object", uri);
-	}
-	WT_ERR_NOTFOUND_OK(ret);
-	if (!target_list)
-		return (0);
-
-	/* Add the hot backup and single-threading file to the list. */
-	WT_ERR(__backup_list_append(session, cb, WT_METADATA_BACKUP));
-	WT_ERR(__backup_list_append(session, cb, WT_SINGLETHREAD));
-
-err:	if (path != NULL)
-		__wt_free(session, path);
-	__wt_scr_free(&tmp);
-	return (ret);
-}
-
-/*
- * __backup_table --
- *	Back up all the elements associated with a table.
- */
-static int
-__backup_table(
-    WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb, const char *name, FILE *bfp)
-{
-	WT_CURSOR *cursor;
-	WT_DECL_RET;
-
-	cursor = NULL;
-
-	/* Open a cursor on the metadata file. */
-	WT_RET(__wt_metadata_cursor(session, NULL, &cursor));
-
-	/* Copy the table's entries... */
-	WT_ERR(__backup_table_element(
-	    session, NULL, cursor, "table", name, bfp));
-	WT_ERR(__backup_table_element(
-	    session, NULL, cursor, "colgroup", name, bfp));
-	WT_ERR(__backup_table_element(
-	    session, NULL, cursor, "index", name, bfp));
-	WT_ERR(__backup_table_element(session, NULL, cursor, "lsm", name, bfp));
-	WT_ERR(__backup_table_element(session, cb, cursor, "file", name, bfp));
-
-err:	if (cursor != NULL)
-		WT_TRET(cursor->close(cursor));
-	return (ret);
-}
-
-/*
- * __backup_table_element --
- *	Backup the column groups or indices.
- */
-static int
-__backup_table_element(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb,
-    WT_CURSOR *cursor, const char *elem, const char *table, FILE *bfp)
-{
-	WT_DECL_RET;
-	WT_DECL_ITEM(tmp);
-	int cmp;
-	const char *key, *value;
-
-	WT_RET(__wt_scr_alloc(session, 512, &tmp));
-
-	WT_ERR(__wt_buf_fmt(session, tmp, "%s:%s", elem, table));
-	cursor->set_key(cursor, tmp->data);
-	if ((ret = cursor->search_near(cursor, &cmp)) == 0 && cmp < 0)
-		ret = cursor->next(cursor);
-	for (; ret == 0; ret = cursor->next(cursor)) {
-		WT_ERR(cursor->get_key(cursor, &key));
-		if (strncmp(tmp->data, key, strlen(tmp->data)) != 0)
-			break;
-
-		/*
-		 * XXX
-		 * This is wrong: we could match non-unique table names, so for
-		 * example, a target of "table_X" will match both "table_X" and
-		 * "table_XX".
-		 */
-
-		/* Dump the metadata entry. */
-		WT_ERR(cursor->get_value(cursor, &value));
-		WT_ERR_TEST(
-		    (fprintf(bfp, "%s\n%s\n", key, value) < 0), __wt_errno());
-
-		/*
-		 * If it's a physical object (file:), add it to our list of
-		 * objects to copy.
-		 */
-		if (cb != NULL)
-			WT_ERR(__backup_list_append(
-			    session, cb, key + strlen(elem) + 1));
+		WT_ERR(__wt_schema_worker(
+		    session, uri, NULL, __wt_backup_list_append, cfg, 0));
 	}
 	WT_ERR_NOTFOUND_OK(ret);
 
@@ -495,16 +375,14 @@ err:	__wt_scr_free(&tmp);
  *	Create the meta-data backup file.
  */
 static int
-__backup_file_create(WT_SESSION_IMPL *session, FILE **fpp)
+__backup_file_create(WT_SESSION_IMPL *session, WT_CURSOR_BACKUP *cb)
 {
 	WT_DECL_RET;
 	const char *path;
 
-	*fpp = NULL;
-
 	/* Open the hot backup file. */
 	WT_RET(__wt_filename(session, WT_METADATA_BACKUP, &path));
-	WT_ERR_TEST((*fpp = fopen(path, "w")) == NULL, __wt_errno());
+	WT_ERR_TEST((cb->bfp = fopen(path, "w")) == NULL, __wt_errno());
 
 err:	__wt_free(session, path);
 	return (ret);
@@ -521,6 +399,32 @@ __backup_file_remove(WT_SESSION_IMPL *session)
 }
 
 /*
+ * __wt_backup_list_append --
+ *	Append a new file name to the list, allocated space as necessary.
+ *	Called via the schema_worker function.
+ */
+int
+__wt_backup_list_append(WT_SESSION_IMPL *session, const char *name)
+{
+	WT_CURSOR_BACKUP *cb;
+	const char *value;
+
+	cb = session->bkp_cursor;
+
+	/* Add the metadata entry to the backup file. */
+	WT_RET(__wt_metadata_read(session, name, &value));
+	WT_RET_TEST(
+	    (fprintf(cb->bfp, "%s\n%s\n", name, value) < 0), __wt_errno());
+
+	/* Add to the list of files needing to be copied. */
+	if (WT_PREFIX_MATCH(name, "file:"))
+		WT_RET(
+		    __backup_list_append(session, cb, name + strlen("file:")));
+
+	return (0);
+}
+
+/*
  * __backup_list_append --
  *	Append a new file name to the list, allocated space as necessary.
  */
@@ -542,7 +446,8 @@ __backup_list_append(
 	 * that for now, that block manager might not even support physical
 	 * copying of files by applications.
 	 */
-	WT_RET(__wt_strdup(session, name, &cb->list[cb->list_next++]));
+	WT_RET(__wt_strdup(
+	    session, name, &cb->list[cb->list_next++]));
 	cb->list[cb->list_next] = NULL;
 
 	return (0);
diff --git a/src/include/cursor.h b/src/include/cursor.h
index 3cc945ca2f3..d1af0aa2798 100644
--- a/src/include/cursor.h
+++ b/src/include/cursor.h
@@ -60,6 +60,7 @@ struct __wt_cursor_backup {
 	size_t list_allocated;		/* List of files */
 	size_t list_next;
 	char **list;
+	FILE *bfp;
 };
 
 struct __wt_cursor_btree {
diff --git a/src/include/extern.h b/src/include/extern.h
index 62645214d4b..6a1ef920c7b 100644
--- a/src/include/extern.h
+++ b/src/include/extern.h
@@ -582,6 +582,7 @@ extern int __wt_curbackup_open(WT_SESSION_IMPL *session,
     const char *uri,
     const char *cfg[],
     WT_CURSOR **cursorp);
+extern int __wt_backup_list_append(WT_SESSION_IMPL *session, const char *name);
 extern int __wt_curbulk_init(WT_CURSOR_BULK *cbulk, int bitmap);
 extern int __wt_curconfig_open(WT_SESSION_IMPL *session,
     const char *uri,
@@ -726,8 +727,10 @@ extern int __wt_lsm_tree_truncate( WT_SESSION_IMPL *session,
     const char *cfg[]);
 extern int __wt_lsm_tree_worker(WT_SESSION_IMPL *session,
     const char *uri,
-    int (*func)(WT_SESSION_IMPL *,
+    int (*file_func)(WT_SESSION_IMPL *,
     const char *[]),
+    int (*name_func)(WT_SESSION_IMPL *,
+    const char *),
     const char *cfg[],
     uint32_t open_flags);
 extern void *__wt_lsm_merge_worker(void *vargs);
@@ -763,7 +766,6 @@ extern void __wt_meta_ckptlist_free(WT_SESSION_IMPL *session,
     WT_CKPT *ckptbase);
 extern void __wt_meta_checkpoint_free(WT_SESSION_IMPL *session, WT_CKPT *ckpt);
 extern int __wt_metadata_open(WT_SESSION_IMPL *session);
-extern int __wt_metadata_load_backup(WT_SESSION_IMPL *session);
 extern int __wt_metadata_cursor( WT_SESSION_IMPL *session,
     const char *config,
     WT_CURSOR **cursorp);
@@ -789,11 +791,11 @@ extern int __wt_meta_track_fileop( WT_SESSION_IMPL *session,
     const char *olduri,
     const char *newuri);
 extern int __wt_meta_track_handle_lock(WT_SESSION_IMPL *session, int created);
-extern int __wt_meta_turtle_init(WT_SESSION_IMPL *session, int *existp);
-extern int __wt_meta_turtle_read( WT_SESSION_IMPL *session,
+extern int __wt_turtle_init(WT_SESSION_IMPL *session);
+extern int __wt_turtle_read(WT_SESSION_IMPL *session,
     const char *key,
     const char **valuep);
-extern int __wt_meta_turtle_update( WT_SESSION_IMPL *session,
+extern int __wt_turtle_update( WT_SESSION_IMPL *session,
     const char *key,
     const char *value);
 extern void __wt_abort(WT_SESSION_IMPL *session) WT_GCC_ATTRIBUTE((noreturn));
@@ -1048,8 +1050,10 @@ extern int __wt_schema_get_source( WT_SESSION_IMPL *session,
 extern int __wt_schema_name_check(WT_SESSION_IMPL *session, const char *uri);
 extern int __wt_schema_worker(WT_SESSION_IMPL *session,
     const char *uri,
-    int (*func)(WT_SESSION_IMPL *,
+    int (*file_func)(WT_SESSION_IMPL *,
     const char *[]),
+    int (*name_func)(WT_SESSION_IMPL *,
+    const char *),
     const char *cfg[],
     uint32_t open_flags);
 extern int __wt_open_cursor(WT_SESSION_IMPL *session,
diff --git a/src/include/session.h b/src/include/session.h
index 8565fa50418..64fbfcd8185 100644
--- a/src/include/session.h
+++ b/src/include/session.h
@@ -62,6 +62,7 @@ struct __wt_session_impl {
 	WT_CURSOR *cursor;		/* Current cursor */
 					/* Cursors closed with the session */
 	TAILQ_HEAD(__cursors, __wt_cursor) cursors;
+	WT_CURSOR_BACKUP *bkp_cursor;	/* Cursor for current backup */
 
 	WT_BTREE *metafile;		/* Metadata file */
 	void	*meta_track;		/* Metadata operation tracking */
diff --git a/src/lsm/lsm_tree.c b/src/lsm/lsm_tree.c
index 1131c2f09f8..9a92cceae3e 100644
--- a/src/lsm/lsm_tree.c
+++ b/src/lsm/lsm_tree.c
@@ -820,7 +820,8 @@ err:	if (locked)
 int
 __wt_lsm_tree_worker(WT_SESSION_IMPL *session,
    const char *uri,
-   int (*func)(WT_SESSION_IMPL *, const char *[]),
+   int (*file_func)(WT_SESSION_IMPL *, const char *[]),
+   int (*name_func)(WT_SESSION_IMPL *, const char *),
    const char *cfg[], uint32_t open_flags)
 {
 	WT_DECL_RET;
@@ -832,11 +833,11 @@ __wt_lsm_tree_worker(WT_SESSION_IMPL *session,
 	    FLD_ISSET(open_flags, WT_DHANDLE_EXCLUSIVE) ? 1 : 0, &lsm_tree));
 	for (i = 0; i < lsm_tree->nchunks; i++) {
 		chunk = lsm_tree->chunk[i];
-		if (func == __wt_checkpoint &&
+		if (file_func == __wt_checkpoint &&
 		    F_ISSET(chunk, WT_LSM_CHUNK_ONDISK))
 			continue;
-		WT_ERR(__wt_schema_worker(
-		    session, chunk->uri, func, cfg, open_flags));
+		WT_ERR(__wt_schema_worker(session, chunk->uri,
+		    file_func, name_func, cfg, open_flags));
 	}
 err:	__wt_lsm_tree_release(session, lsm_tree);
 	return (ret);
diff --git a/src/lsm/lsm_worker.c b/src/lsm/lsm_worker.c
index ae6353197db..e29607f9901 100644
--- a/src/lsm/lsm_worker.c
+++ b/src/lsm/lsm_worker.c
@@ -279,7 +279,7 @@ __wt_lsm_checkpoint_worker(void *arg)
 			F_SET(lsm_tree, WT_LSM_TREE_LOCKED);
 			WT_WITH_SCHEMA_LOCK(session,
 			    ret = __wt_schema_worker(session, chunk->uri,
-			    __wt_checkpoint, NULL, 0));
+			    __wt_checkpoint, NULL, NULL, 0));
 			F_CLR(lsm_tree, WT_LSM_TREE_LOCKED);
 
 			if (ret != 0) {
@@ -432,11 +432,19 @@ __lsm_discard_handle(
 static int
 __lsm_drop_file(WT_SESSION_IMPL *session, const char *uri)
 {
+	WT_CONNECTION_IMPL *conn;
 	WT_DECL_RET;
+	int hot_backup_locked;
 	const char *drop_cfg[] = {
 	    WT_CONFIG_BASE(session, session_drop), "remove_files=false", NULL
 	};
 
+	conn = S2C(session);
+	hot_backup_locked = 0;
+	/* Give up if a hot backup is in progress. */
+	if (conn->hot_backup != 0)
+		return (EBUSY);
+
 	/*
 	 * We need to grab the schema lock to drop the file, so first try to
 	 * make sure there is minimal work to freeing space in the cache.
@@ -445,12 +453,18 @@ __lsm_drop_file(WT_SESSION_IMPL *session, const char *uri)
 	WT_RET(__lsm_discard_handle(session, uri, NULL));
 	WT_RET(__lsm_discard_handle(session, uri, "WiredTigerCheckpoint"));
 
+	__wt_spin_lock(session, &conn->hot_backup_lock);
+	hot_backup_locked = 1;
+	if (conn->hot_backup != 0)
+		goto done;
 	WT_WITH_SCHEMA_LOCK(session,
 	    ret = __wt_schema_drop(session, uri, drop_cfg));
 
 	if (ret == 0)
 		ret = __wt_remove(session, uri + strlen("file:"));
 
+done:	if (hot_backup_locked)
+		__wt_spin_unlock(session, &conn->hot_backup_lock);
 	return (ret);
 }
 
diff --git a/src/meta/meta_table.c b/src/meta/meta_table.c
index 3bcf8e588db..388eb484235 100644
--- a/src/meta/meta_table.c
+++ b/src/meta/meta_table.c
@@ -49,54 +49,6 @@ __wt_metadata_open(WT_SESSION_IMPL *session)
 }
 
 /*
- * __wt_metadata_load_backup --
- *	Load the contents of any hot backup file.
- */
-int
-__wt_metadata_load_backup(WT_SESSION_IMPL *session)
-{
-	FILE *fp;
-	WT_DECL_ITEM(key);
-	WT_DECL_ITEM(value);
-	WT_DECL_RET;
-	const char *path;
-
-	fp = NULL;
-	path = NULL;
-
-	/* Look for a hot backup file: if we find it, load it. */
-	WT_RET(__wt_filename(session, WT_METADATA_BACKUP, &path));
-	if ((fp = fopen(path, "r")) == NULL) {
-		__wt_free(session, path);
-		return (0);
-	}
-
-	/* Read line pairs and load them into the metadata file. */
-	WT_ERR(__wt_scr_alloc(session, 512, &key));
-	WT_ERR(__wt_scr_alloc(session, 512, &value));
-	for (;;) {
-		WT_ERR(__wt_getline(session, key, fp));
-		if (key->size == 0)
-			break;
-		WT_ERR(__wt_getline(session, value, fp));
-		if (value->size == 0)
-			WT_ERR(__wt_illegal_value(session, WT_METADATA_BACKUP));
-		WT_ERR(__wt_metadata_update(session, key->data, value->data));
-	}
-
-	/* Remove the hot backup file, it's only read (successfully) once. */
-	WT_ERR(__wt_remove(session, WT_METADATA_BACKUP));
-
-err:	if (fp != NULL)
-		WT_TRET(fclose(fp) == 0 ? 0 : __wt_errno());
-	if (path != NULL)
-		__wt_free(session, path);
-	__wt_scr_free(&key);
-	__wt_scr_free(&value);
-	return (ret);
-}
-
-/*
  * __wt_metadata_cursor --
  *	Opens a cursor on the metadata.
  */
@@ -159,7 +111,7 @@ __wt_metadata_update(
 	WT_DECL_RET;
 
 	if (__metadata_turtle(key))
-		return (__wt_meta_turtle_update(session, key, value));
+		return (__wt_turtle_update(session, key, value));
 
 	if (WT_META_TRACKING(session))
 		WT_RET(__wt_meta_track_update(session, key));
@@ -214,7 +166,7 @@ __wt_metadata_read(
 	*valuep = NULL;
 
 	if (__metadata_turtle(key))
-		return (__wt_meta_turtle_read(session, key, valuep));
+		return (__wt_turtle_read(session, key, valuep));
 
 	WT_RET(__wt_metadata_cursor(session, NULL, &cursor));
 	cursor->set_key(cursor, key);
diff --git a/src/meta/meta_turtle.c b/src/meta/meta_turtle.c
index fedd0ff01f8..229431d296d 100644
--- a/src/meta/meta_turtle.c
+++ b/src/meta/meta_turtle.c
@@ -8,56 +8,162 @@
 #include "wt_internal.h"
 
 /*
- * __wt_meta_turtle_init --
+ * __metadata_config --
+ *	Return the default configuration information for the metadata file.
+ */
+static int
+__metadata_config(WT_SESSION_IMPL *session, const char **metaconfp)
+{
+	WT_DECL_RET;
+	WT_ITEM *buf;
+	const char *cfg[] = { WT_CONFIG_BASE(session, file_meta), NULL, NULL };
+	const char *metaconf;
+
+	*metaconfp = NULL;
+
+	buf = NULL;
+	metaconf = NULL;
+
+	/* Create a turtle file with default values. */
+	WT_RET(__wt_scr_alloc(session, 0, &buf));
+	WT_ERR(__wt_buf_fmt(session, buf,
+	    "key_format=S,value_format=S,version=(major=%d,minor=%d)",
+	    WT_BTREE_MAJOR_VERSION, WT_BTREE_MINOR_VERSION));
+	cfg[1] = buf->data;
+	WT_ERR(__wt_config_collapse(session, cfg, &metaconf));
+
+	*metaconfp = metaconf;
+
+	if (0) {
+err:		__wt_free(session, metaconf);
+	}
+	__wt_scr_free(&buf);
+	return (ret);
+}
+
+/*
+ * __metadata_init --
+ *	Create the metadata file.
+ */
+static int
+__metadata_init(WT_SESSION_IMPL *session)
+{
+	WT_DECL_RET;
+
+	/*
+	 * We're single-threaded, but acquire the schema lock regardless: the
+	 * lower level code checks that it is appropriately synchronized.
+	 */
+	WT_WITH_SCHEMA_LOCK(session,
+	    ret = __wt_schema_create(session, WT_METADATA_URI, NULL));
+
+	return (ret);
+}
+
+/*
+ * __metadata_load_hot_backup --
+ *	Load the contents of any hot backup file.
+ */
+static int
+__metadata_load_hot_backup(WT_SESSION_IMPL *session)
+{
+	FILE *fp;
+	WT_DECL_ITEM(key);
+	WT_DECL_ITEM(value);
+	WT_DECL_RET;
+	const char *path;
+
+	fp = NULL;
+	path = NULL;
+
+	/* Look for a hot backup file: if we find it, load it. */
+	WT_RET(__wt_filename(session, WT_METADATA_BACKUP, &path));
+	fp = fopen(path, "r");
+	__wt_free(session, path);
+	if (fp == NULL)
+		return (0);
+
+	/* Read line pairs and load them into the metadata file. */
+	WT_ERR(__wt_scr_alloc(session, 512, &key));
+	WT_ERR(__wt_scr_alloc(session, 512, &value));
+	for (;;) {
+		WT_ERR(__wt_getline(session, key, fp));
+		if (key->size == 0)
+			break;
+		WT_ERR(__wt_getline(session, value, fp));
+		if (value->size == 0)
+			WT_ERR(__wt_illegal_value(session, WT_METADATA_BACKUP));
+		WT_ERR(__wt_metadata_update(session, key->data, value->data));
+	}
+
+err:	if (fp != NULL)
+		WT_TRET(fclose(fp) == 0 ? 0 : __wt_errno());
+	__wt_scr_free(&key);
+	__wt_scr_free(&value);
+	return (ret);
+}
+
+/*
+ * __wt_turtle_init --
  *	Check the turtle file and create if necessary.
  */
 int
-__wt_meta_turtle_init(WT_SESSION_IMPL *session, int *existp)
+__wt_turtle_init(WT_SESSION_IMPL *session)
 {
 	WT_DECL_RET;
-	WT_ITEM *buf;
 	int exist;
-	const char *cfg[3], *metaconf;
+	const char *metaconf;
 
-	buf = NULL;
 	metaconf = NULL;
-	*existp = 0;
 
-	/* Discard any turtle setup file left-over from previous runs. */
+	/*
+	 * Discard any turtle setup file left-over from previous runs.  This
+	 * doesn't matter for correctness, it's just cleaning up random files.
+	 */
 	WT_RET(__wt_exist(session, WT_METADATA_TURTLE_SET, &exist));
 	if (exist)
 		WT_RET(__wt_remove(session, WT_METADATA_TURTLE_SET));
 
-	/* If there's already a turtle file, we're done. */
+	/*
+	 * We could die after creating the turtle file and before creating the
+	 * metadata file, or worse, the metadata file might be in some random
+	 * state.  Make sure that doesn't happen: if we don't find the turtle
+	 * file, first create the metadata file, load any hot backup, and then
+	 * create the turtle file.  No matter what happens, if metadata file
+	 * creation doesn't fully complete, we won't have a turtle file and we
+	 * will repeat the process until we succeed.
+	 *
+	 * If there's already a turtle file, we're done.
+	 */
 	WT_RET(__wt_exist(session, WT_METADATA_TURTLE, &exist));
-	if (exist) {
-		*existp = 1;
+	if (exist)
 		return (0);
-	}
 
-	/* Create a turtle file with default values. */
-	WT_ERR(__wt_scr_alloc(session, 0, &buf));
-	WT_ERR(__wt_buf_fmt(session, buf,
-	    "key_format=S,value_format=S,version=(major=%d,minor=%d)",
-	    WT_BTREE_MAJOR_VERSION, WT_BTREE_MINOR_VERSION));
-	cfg[0] = WT_CONFIG_BASE(session, file_meta);
-	cfg[1] = buf->data;
-	cfg[2] = NULL;
-	WT_ERR(__wt_config_collapse(session, cfg, &metaconf));
-	WT_ERR(__wt_meta_turtle_update(session, WT_METADATA_URI, metaconf));
+	/* Create the metadata file. */
+	WT_RET(__metadata_init(session));
+
+	/* Load any hot-backup information. */
+	WT_RET(__metadata_load_hot_backup(session));
+
+	/* Create the turtle file. */
+	WT_RET(__metadata_config(session, &metaconf));
+	WT_ERR(__wt_turtle_update(session, WT_METADATA_URI, metaconf));
+
+	/* Remove the backup file if it exists, we'll never read it again. */
+	WT_ERR(__wt_exist(session, WT_METADATA_BACKUP, &exist));
+	if (exist)
+		WT_ERR(__wt_remove(session, WT_METADATA_BACKUP));
 
 err:	__wt_free(session, metaconf);
-	__wt_scr_free(&buf);
 	return (ret);
 }
 
 /*
- * __wt_meta_turtle_read --
+ * __wt_turtle_read --
  *	Read the turtle file.
  */
 int
-__wt_meta_turtle_read(
-    WT_SESSION_IMPL *session, const char *key, const char **valuep)
+__wt_turtle_read(WT_SESSION_IMPL *session, const char *key, const char **valuep)
 {
 	FILE *fp;
 	WT_DECL_ITEM(buf);
@@ -70,9 +176,19 @@ __wt_meta_turtle_read(
 	fp = NULL;
 	path = NULL;
 
-	/* Open the turtle file. */
+	/*
+	 * Open the turtle file; there's one case where we won't find the turtle
+	 * file, yet still succeed.  We create the metadata file before creating
+	 * the turtle file, and that means returning the default configuration
+	 * string for the metadata file.
+	 */
 	WT_RET(__wt_filename(session, WT_METADATA_TURTLE, &path));
-	WT_ERR_TEST((fp = fopen(path, "r")) == NULL, __wt_errno());
+	if ((fp = fopen(path, "r")) == NULL)
+		ret = __wt_errno();
+	__wt_free(session, path);
+	if (fp == NULL)
+		return (strcmp(key, WT_METADATA_URI) == 0 ?
+		    __metadata_config(session, valuep) : ret);
 
 	/* Search for the key. */
 	WT_ERR(__wt_scr_alloc(session, 512, &buf));
@@ -96,38 +212,38 @@ __wt_meta_turtle_read(
 
 err:	if (fp != NULL)
 		WT_TRET(fclose(fp) == 0 ? 0 : __wt_errno());
-	if (path != NULL)
-		__wt_free(session, path);
 	__wt_scr_free(&buf);
 	return (ret);
 }
 
 /*
- * __wt_meta_turtle_update --
+ * __wt_turtle_update --
  *	Update the turtle file.
  */
 int
-__wt_meta_turtle_update(
+__wt_turtle_update(
     WT_SESSION_IMPL *session, const char *key,  const char *value)
 {
 	FILE *fp;
 	WT_DECL_RET;
-	WT_ITEM *buf;
 	int vmajor, vminor, vpatch;
 	const char *path, *version;
 
-	buf = NULL;
 	fp = NULL;
-
-	version = wiredtiger_version(&vmajor, &vminor, &vpatch);
+	path = NULL;
 
 	/*
 	 * Create the turtle setup file: we currently re-write it from scratch
 	 * every time.
 	 */
-	WT_ERR(__wt_filename(session, WT_METADATA_TURTLE_SET, &path));
-	WT_ERR_TEST((fp = fopen(path, "w")) == NULL, __wt_errno());
+	WT_RET(__wt_filename(session, WT_METADATA_TURTLE_SET, &path));
+	if ((fp = fopen(path, "w")) == NULL)
+		ret = __wt_errno();
+	__wt_free(session, path);
+	if (fp == NULL)
+		return (ret);
 
+	version = wiredtiger_version(&vmajor, &vminor, &vpatch);
 	WT_ERR_TEST((fprintf(fp,
 	    "%s\n%s\n%s\n" "major=%d,minor=%d,patch=%d\n%s\n%s\n",
 	    WT_METADATA_VERSION_STR, version,
@@ -147,8 +263,5 @@ err:		WT_TRET(__wt_remove(session, WT_METADATA_TURTLE_SET));
 
 	if  (fp != NULL)
 		WT_TRET(fclose(fp) == 0 ? 0 : __wt_errno());
-	__wt_free(session, path);
-	__wt_scr_free(&buf);
-
 	return (ret);
 }
diff --git a/src/schema/schema_worker.c b/src/schema/schema_worker.c
index 8546dff84af..485c751acd7 100644
--- a/src/schema/schema_worker.c
+++ b/src/schema/schema_worker.c
@@ -15,7 +15,8 @@
 int
 __wt_schema_worker(WT_SESSION_IMPL *session,
    const char *uri,
-   int (*func)(WT_SESSION_IMPL *, const char *[]),
+   int (*file_func)(WT_SESSION_IMPL *, const char *[]),
+   int (*name_func)(WT_SESSION_IMPL *, const char *),
    const char *cfg[], uint32_t open_flags)
 {
 	WT_COLGROUP *colgroup;
@@ -30,50 +31,65 @@ __wt_schema_worker(WT_SESSION_IMPL *session,
 	table = NULL;
 	tablename = uri;
 
+	if (name_func != NULL)
+		WT_ERR(name_func(session, uri));
+
 	/* Get the btree handle(s) and call the underlying function. */
 	if (WT_PREFIX_MATCH(uri, "file:")) {
-		WT_ERR(__wt_session_get_btree_ckpt(
-		    session, uri, cfg, open_flags));
-		ret = func(session, cfg);
-		WT_TRET(__wt_session_release_btree(session));
+		if (file_func != NULL) {
+			WT_ERR(__wt_session_get_btree_ckpt(
+			    session, uri, cfg, open_flags));
+			ret = file_func(session, cfg);
+			WT_TRET(__wt_session_release_btree(session));
+		}
 	} else if (WT_PREFIX_MATCH(uri, "colgroup:")) {
 		WT_ERR(__wt_schema_get_colgroup(session, uri, NULL, &colgroup));
-		WT_ERR(__wt_schema_worker(
-		    session, colgroup->source, func, cfg, open_flags));
+		WT_ERR(__wt_schema_worker(session, colgroup->source,
+		    file_func, name_func, cfg, open_flags));
 	} else if (WT_PREFIX_SKIP(tablename, "index:")) {
 		idx = NULL;
 		WT_ERR(__wt_schema_get_index(session, uri, NULL, &idx));
-		WT_ERR(__wt_schema_worker(
-		    session, idx->source, func, cfg, open_flags));
+		WT_ERR(__wt_schema_worker(session, idx->source,
+		    file_func, name_func, cfg, open_flags));
 	} else if (WT_PREFIX_MATCH(uri, "lsm:")) {
 		WT_ERR(__wt_lsm_tree_worker(
-		    session, uri, func, cfg, open_flags));
+		    session, uri, file_func, name_func, cfg, open_flags));
 	} else if (WT_PREFIX_SKIP(tablename, "table:")) {
 		WT_ERR(__wt_schema_get_table(session,
 		    tablename, strlen(tablename), 0, &table));
 		WT_ASSERT(session, session->dhandle == NULL);
 
+		/*
+		 * We could make a recursive call for each colgroup or index
+		 * URI, but since we have already opened the table, we can take
+		 * a short cut and skip straight to the sources.  If we have a
+		 * name function, it needs to know about the intermediate URIs.
+		 */
 		for (i = 0; i < WT_COLGROUPS(table); i++) {
 			colgroup = table->cgroups[i];
-			WT_ERR(__wt_schema_worker(
-			    session, colgroup->source, func, cfg, open_flags));
+			if (name_func != NULL)
+				WT_ERR(name_func(session, colgroup->name));
+			WT_ERR(__wt_schema_worker(session, colgroup->source,
+			    file_func, name_func, cfg, open_flags));
 		}
 
 		WT_ERR(__wt_schema_open_indices(session, table));
 		for (i = 0; i < table->nindices; i++) {
 			idx = table->indices[i];
-			WT_ERR(__wt_schema_worker(
-			    session, idx->source, func, cfg, open_flags));
+			if (name_func != NULL)
+				WT_ERR(name_func(session, idx->name));
+			WT_ERR(__wt_schema_worker(session, idx->source,
+			    file_func, name_func, cfg, open_flags));
 		}
 	} else if ((ret = __wt_schema_get_source(session, uri, &dsrc)) == 0) {
 		wt_session = (WT_SESSION *)session;
-		if (func == __wt_compact && dsrc->compact != NULL)
+		if (file_func == __wt_compact && dsrc->compact != NULL)
 			WT_ERR(dsrc->compact(
 			    dsrc, wt_session, uri, (WT_CONFIG_ARG *)cfg));
-		else if (func == __wt_salvage && dsrc->salvage != NULL)
+		else if (file_func == __wt_salvage && dsrc->salvage != NULL)
 			WT_ERR(dsrc->salvage(
 			   dsrc, wt_session, uri, (WT_CONFIG_ARG *)cfg));
-		else if (func == __wt_verify && dsrc->verify != NULL)
+		else if (file_func == __wt_verify && dsrc->verify != NULL)
 			WT_ERR(dsrc->verify(
 			   dsrc, wt_session, uri, (WT_CONFIG_ARG *)cfg));
 		else
diff --git a/src/session/session_api.c b/src/session/session_api.c
index 2d037051955..68d04a8f640 100644
--- a/src/session/session_api.c
+++ b/src/session/session_api.c
@@ -350,7 +350,7 @@ __session_compact_worker(
 	SESSION_API_CALL(session, compact, config, cfg);
 
 	WT_WITH_SCHEMA_LOCK(session,
-	    ret = __wt_schema_worker(session, uri, __wt_compact, cfg, 0));
+	    ret = __wt_schema_worker(session, uri, __wt_compact, NULL, cfg, 0));
 
 err:	API_END_NOTFOUND_MAP(session, ret);
 }
@@ -466,8 +466,8 @@ __session_salvage(WT_SESSION *wt_session, const char *uri, const char *config)
 
 	SESSION_API_CALL(session, salvage, config, cfg);
 	WT_WITH_SCHEMA_LOCK(session,
-	    ret = __wt_schema_worker(session, uri,
-		__wt_salvage, cfg, WT_DHANDLE_EXCLUSIVE | WT_BTREE_SALVAGE));
+	    ret = __wt_schema_worker(session, uri, __wt_salvage,
+		NULL, cfg, WT_DHANDLE_EXCLUSIVE | WT_BTREE_SALVAGE));
 
 err:	API_END_NOTFOUND_MAP(session, ret);
 }
@@ -588,8 +588,8 @@ __session_upgrade(WT_SESSION *wt_session, const char *uri, const char *config)
 
 	SESSION_API_CALL(session, upgrade, config, cfg);
 	WT_WITH_SCHEMA_LOCK(session,
-	    ret = __wt_schema_worker(session, uri,
-		__wt_upgrade, cfg, WT_DHANDLE_EXCLUSIVE | WT_BTREE_UPGRADE));
+	    ret = __wt_schema_worker(session, uri, __wt_upgrade,
+		NULL, cfg, WT_DHANDLE_EXCLUSIVE | WT_BTREE_UPGRADE));
 
 err:	API_END_NOTFOUND_MAP(session, ret);
 }
@@ -608,8 +608,8 @@ __session_verify(WT_SESSION *wt_session, const char *uri, const char *config)
 
 	SESSION_API_CALL(session, verify, config, cfg);
 	WT_WITH_SCHEMA_LOCK(session,
-	    ret = __wt_schema_worker(session, uri,
-		__wt_verify, cfg, WT_DHANDLE_EXCLUSIVE | WT_BTREE_VERIFY));
+	    ret = __wt_schema_worker(session, uri, __wt_verify,
+		NULL, cfg, WT_DHANDLE_EXCLUSIVE | WT_BTREE_VERIFY));
 
 err:	API_END_NOTFOUND_MAP(session, ret);
 }
diff --git a/src/txn/txn_ckpt.c b/src/txn/txn_ckpt.c
index 8d4b56de2f1..d1c14362758 100644
--- a/src/txn/txn_ckpt.c
+++ b/src/txn/txn_ckpt.c
@@ -95,7 +95,7 @@ __checkpoint_apply(WT_SESSION_IMPL *session, const char *cfg[],
 
 		WT_ERR(__wt_buf_fmt(session, tmp, "%.*s", (int)k.len, k.str));
 		if ((ret = __wt_schema_worker(
-		    session, tmp->data, op, cfg, 0)) != 0)
+		    session, tmp->data, op, NULL, cfg, 0)) != 0)
 			WT_ERR_MSG(session, ret, "%s", (const char *)tmp->data);
 	}
 	WT_ERR_NOTFOUND_OK(ret);
diff --git a/test/suite/test_backup.py b/test/suite/test_backup.py
index dbd40d45d71..0ab7ab2e5ec 100644
--- a/test/suite/test_backup.py
+++ b/test/suite/test_backup.py
@@ -140,14 +140,6 @@ class test_backup(wttest.WiredTigerTestCase, suite_subprocess):
         self.backup_table([3,4,5])
         self.backup_table([5,6,7])
 
-    # Test backup of random object types fails.
-    def test_illegal_objects(self):
-        for target in ('colgroup:xxx', 'index:xxx'):
-            msg = '/unsupported backup target object/'
-            config = 'target=("%s")' % target
-            self.assertRaisesWithMessage(wiredtiger.WiredTigerError,
-                lambda: self.session.open_cursor('backup:', None, config), msg)
-
     # Test cursor reset runs through the list twice.
     def test_cursor_reset(self):
         self.populate(0)
diff --git a/test/suite/test_backup03.py b/test/suite/test_backup03.py
new file mode 100644
index 00000000000..a8b9340cba2
--- /dev/null
+++ b/test/suite/test_backup03.py
@@ -0,0 +1,128 @@
+#!/usr/bin/env python
+#
+# Public Domain 2008-2013 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 glob
+import os
+import shutil
+import string
+from suite_subprocess import suite_subprocess
+import wiredtiger, wttest
+from helper import compare_files,\
+    complex_populate, complex_populate_lsm, simple_populate
+
+# test_backup03.py
+#    Utilities: wt backup
+# Test cursor backup with target URIs
+class test_backup(wttest.WiredTigerTestCase, suite_subprocess):
+    dir='backup.dir'            # Backup directory name
+
+    pfx = 'test_backup'
+    objs = [
+        ('table:' + pfx + '.1',  simple_populate, 100),
+        ('lsm:' + pfx + '.2',  simple_populate, 50000),
+        ('table:' + pfx + '.3', complex_populate, 100),
+        ('table:' + pfx + '.4', complex_populate_lsm, 100),
+    ]
+
+    # Populate a set of objects.
+    def populate(self):
+        for i in self.objs:
+            i[1](self, i[0], 'key_format=S', i[2])
+        # Backup needs a checkpoint
+        self.session.checkpoint(None)
+
+    # Compare the original and backed-up files using the wt dump command.
+    def compare(self, uri):
+        self.runWt(['dump', uri], outfilename='orig')
+        self.runWt(['-h', self.dir, 'dump', uri], outfilename='backup')
+        compare_files(self, 'orig', 'backup')
+
+    # Check that a URI doesn't exist, both the meta-data and the file names.
+    def confirmPathDoesNotExist(self, uri):
+        conn = wiredtiger.wiredtiger_open(self.dir)
+        session = conn.open_session()
+        self.assertRaises(wiredtiger.WiredTigerError,
+            lambda: session.open_cursor(uri, None, None))
+        conn.close()
+
+        self.assertEqual(
+            glob.glob(self.dir + '*' + uri.split(":")[1] + '*'), [],
+            'confirmPathDoesNotExist: URI exists, file name matching \"' +
+            uri.split(":")[1] + '\" found')
+
+    # Backup a set of chosen tables/files using the wt backup command.
+    def backup_table_cursor(self, l):
+        # Remove any previous backup directories.
+        shutil.rmtree(self.dir, True)
+        os.mkdir(self.dir)
+
+        # Build the target list.
+        config = 'target=('
+        for i in range(0, len(self.objs)):
+            if i in l:
+                config += '"' + self.objs[i][0] + '",'
+        config += ')'
+
+        # Open up the backup cursor, and copy the files.
+        cursor = self.session.open_cursor('backup:', None, config)
+        while True:
+            ret = cursor.next()
+            if ret != 0:
+                break
+            #print 'Copy from: ' + cursor.get_key() + ' to ' + self.dir
+            shutil.copy(cursor.get_key(), self.dir)
+        self.assertEqual(ret, wiredtiger.WT_NOTFOUND)
+        cursor.close()
+
+        # Confirm the objects we backed up exist, with correct contents.
+        for i in range(0, len(self.objs)):
+            if i in l:
+                self.compare(self.objs[i][0])
+
+        # Confirm the other objects don't exist.
+        for i in range(0, len(self.objs)):
+            if i not in l:
+                self.confirmPathDoesNotExist(self.objs[i][0])
+
+    # Test backup with targets
+    def test_targets_groups(self):
+        self.populate()
+        self.backup_table_cursor([0,2])
+        self.backup_table_cursor([1,3])
+        self.backup_table_cursor([0,1,2])
+        self.backup_table_cursor([0,1,2,3])
+
+    def test_target_individual(self):
+        self.populate()
+        self.backup_table_cursor([0])
+        self.backup_table_cursor([1])
+        self.backup_table_cursor([2])
+        self.backup_table_cursor([3])
+
+
+if __name__ == '__main__':
+    wttest.run()