WT-1924 Ensure the metadata is flushed when closing an LSM bulk cursor.

(cherry picked from commit dde4092ebb9bcaede1d550ed32dabe3062272bea)

5 files changed, 44 insertions, 11 deletions

diff --git a/src/lsm/lsm_cursor.c b/src/lsm/lsm_cursor.c
index 56126a4b724..111de7a2be1 100644
--- a/src/lsm/lsm_cursor.c
+++ b/src/lsm/lsm_cursor.c
@@ -1533,6 +1533,7 @@ __wt_clsm_open(WT_SESSION_IMPL *session,
 	if (bulk && (ret == EBUSY || (ret == 0 &&  lsm_tree->nchunks > 1)))
 		WT_ERR_MSG(session, EINVAL,
 		    "bulk-load is only supported on newly created LSM trees");
+	WT_ASSERT(session, !bulk || lsm_tree->exclusive);
 	/* Flag any errors from the tree get. */
 	WT_RET(ret);
 
diff --git a/src/lsm/lsm_cursor_bulk.c b/src/lsm/lsm_cursor_bulk.c
index 6b51a070e47..8099c87c3bf 100644
--- a/src/lsm/lsm_cursor_bulk.c
+++ b/src/lsm/lsm_cursor_bulk.c
@@ -16,13 +16,28 @@ static int
 __clsm_close_bulk(WT_CURSOR *cursor)
 {
 	WT_CURSOR_LSM *clsm;
+	WT_CURSOR *bulk_cursor;
 	WT_LSM_TREE *lsm_tree;
+	WT_SESSION_IMPL *session;
 
 	clsm = (WT_CURSOR_LSM *)cursor;
 	lsm_tree = clsm->lsm_tree;
+	session = (WT_SESSION_IMPL *)clsm->iface.session;
+
+	/* Close the bulk cursor to ensure the chunk is written to disk. */
+	bulk_cursor = clsm->cursors[0];
+	WT_RET(bulk_cursor->close(bulk_cursor));
+	clsm->cursors[0] = NULL;
+	clsm->nchunks = 0;
+
+	/* Set ondisk, and flush the metadata */
 	F_SET(lsm_tree->chunk[0], WT_LSM_CHUNK_ONDISK);
+	WT_RET(__wt_lsm_meta_write(session, lsm_tree));
+	++lsm_tree->dsk_gen;
 
+	/* Close the LSM cursor */
 	WT_RET(__wt_clsm_close(cursor));
+
 	return (0);
 }
 /*
diff --git a/src/lsm/lsm_tree.c b/src/lsm/lsm_tree.c
index 2bded10cb96..63f19858279 100644
--- a/src/lsm/lsm_tree.c
+++ b/src/lsm/lsm_tree.c
@@ -10,7 +10,8 @@
 
 static int __lsm_tree_cleanup_old(WT_SESSION_IMPL *, const char *);
 static int __lsm_tree_open_check(WT_SESSION_IMPL *, WT_LSM_TREE *);
-static int __lsm_tree_open(WT_SESSION_IMPL *, const char *, WT_LSM_TREE **);
+static int __lsm_tree_open(
+    WT_SESSION_IMPL *, const char *, int, WT_LSM_TREE **);
 static int __lsm_tree_set_name(WT_SESSION_IMPL *, WT_LSM_TREE *, const char *);
 
 /*
@@ -430,7 +431,7 @@ __wt_lsm_tree_create(WT_SESSION_IMPL *session,
 	 */
 	if (ret == 0)
 		WT_WITH_DHANDLE_LOCK(session,
-		    ret = __lsm_tree_open(session, uri, &lsm_tree));
+		    ret = __lsm_tree_open(session, uri, 1, &lsm_tree));
 	if (ret == 0)
 		__wt_lsm_tree_release(session, lsm_tree);
 
@@ -539,8 +540,8 @@ __lsm_tree_open_check(WT_SESSION_IMPL *session, WT_LSM_TREE *lsm_tree)
  *	Open an LSM tree structure.
  */
 static int
-__lsm_tree_open(
-    WT_SESSION_IMPL *session, const char *uri, WT_LSM_TREE **treep)
+__lsm_tree_open(WT_SESSION_IMPL *session,
+    const char *uri, int exclusive, WT_LSM_TREE **treep)
 {
 	WT_CONNECTION_IMPL *conn;
 	WT_DECL_RET;
@@ -556,7 +557,8 @@ __lsm_tree_open(
 		WT_RET(__wt_lsm_manager_start(session));
 
 	/* Make sure no one beat us to it. */
-	if ((ret = __lsm_tree_find(session, uri, 0, treep)) != WT_NOTFOUND)
+	if ((ret = __lsm_tree_find(
+	    session, uri, exclusive, treep)) != WT_NOTFOUND)
 		return (ret);
 
 	/* Try to open the tree. */
@@ -582,6 +584,7 @@ __lsm_tree_open(
 	 * with getting handles exclusive.
 	 */
 	lsm_tree->refcnt = 1;
+	lsm_tree->exclusive = exclusive;
 	lsm_tree->queue_ref = 0;
 
 	/* Set a flush timestamp as a baseline. */
@@ -613,8 +616,9 @@ __wt_lsm_tree_get(WT_SESSION_IMPL *session,
 
 	ret = __lsm_tree_find(session, uri, exclusive, treep);
 	if (ret == WT_NOTFOUND)
-		ret = __lsm_tree_open(session, uri, treep);
+		ret = __lsm_tree_open(session, uri, exclusive, treep);
 
+	WT_ASSERT(session, ret != 0 || exclusive == (*treep)->exclusive);
 	return (ret);
 }
 
diff --git a/src/schema/schema_drop.c b/src/schema/schema_drop.c
index 2be8088f83b..694d07c65bf 100644
--- a/src/schema/schema_drop.c
+++ b/src/schema/schema_drop.c
@@ -120,8 +120,13 @@ __drop_table(
 	for (i = 0; i < WT_COLGROUPS(table); i++) {
 		if ((colgroup = table->cgroups[i]) == NULL)
 			continue;
-		WT_ERR(__wt_metadata_remove(session, colgroup->name));
+		/*
+		 * Drop the column group before updating the metadata to avoid
+		 * the metadata for the table becoming inconsistent if we can't
+		 * get exclusive access.
+		 */
 		WT_ERR(__wt_schema_drop(session, colgroup->source, cfg));
+		WT_ERR(__wt_metadata_remove(session, colgroup->name));
 	}
 
 	/* Drop the indices. */
@@ -129,8 +134,13 @@ __drop_table(
 	for (i = 0; i < table->nindices; i++) {
 		if ((idx = table->indices[i]) == NULL)
 			continue;
-		WT_ERR(__wt_metadata_remove(session, idx->name));
+		/*
+		 * Drop the column group before updating the metadata to avoid
+		 * the metadata for the table becoming inconsistent if we can't
+		 * get exclusive access.
+		 */
 		WT_ERR(__wt_schema_drop(session, idx->source, cfg));
+		WT_ERR(__wt_metadata_remove(session, idx->name));
 	}
 
 	WT_ERR(__wt_schema_remove_table(session, table));
diff --git a/src/schema/schema_rename.c b/src/schema/schema_rename.c
index 51281eccec5..c00ffa7d61c 100644
--- a/src/schema/schema_rename.c
+++ b/src/schema/schema_rename.c
@@ -155,15 +155,18 @@ __rename_tree(WT_SESSION_IMPL *session,
 	    cval.str + cval.len));
 
 	/*
+	 * Do the rename before updating the metadata to avoid leaving the
+	 * metadata inconsistent if the rename fails.
+	 */
+	WT_ERR(__wt_schema_rename(session, os->data, ns->data, cfg));
+
+	/*
 	 * Remove the old metadata entry.
 	 * Insert the new metadata entry.
 	 */
 	WT_ERR(__wt_metadata_remove(session, name));
 	WT_ERR(__wt_metadata_insert(session, nn->data, nv->data));
 
-	/* Rename the file. */
-	WT_ERR(__wt_schema_rename(session, os->data, ns->data, cfg));
-
 err:	__wt_scr_free(session, &nn);
 	__wt_scr_free(session, &ns);
 	__wt_scr_free(session, &nv);