path: root/src/lsm/lsm_cursor.c

/*-
 * Copyright (c) 2014-2015 MongoDB, Inc.
 * Copyright (c) 2008-2014 WiredTiger, Inc.
 *	All rights reserved.
 *
 * See the file LICENSE for redistribution information.
 */

#include "wt_internal.h"

#define	WT_FORALL_CURSORS(clsm, c, i)					\
	for ((i) = (clsm)->nchunks; (i) > 0;)				\
		if (((c) = (clsm)->cursors[--i]) != NULL)

#define	WT_LSM_CURCMP(s, lsm_tree, c1, c2, cmp)				\
	__wt_compare(s, (lsm_tree)->collator, &(c1)->key, &(c2)->key, &cmp)

static int __clsm_lookup(WT_CURSOR_LSM *, WT_ITEM *);
static int __clsm_open_cursors(WT_CURSOR_LSM *, bool, u_int, uint32_t);
static int __clsm_reset_cursors(WT_CURSOR_LSM *, WT_CURSOR *);

/*
 * __wt_clsm_request_switch --
 *	Request an LSM tree switch for a cursor operation.
 */
int
__wt_clsm_request_switch(WT_CURSOR_LSM *clsm)
{
	WT_DECL_RET;
	WT_LSM_TREE *lsm_tree;
	WT_SESSION_IMPL *session;

	lsm_tree = clsm->lsm_tree;
	session = (WT_SESSION_IMPL *)clsm->iface.session;

	if (!F_ISSET(lsm_tree, WT_LSM_TREE_NEED_SWITCH)) {
		/*
		 * Check that we are up-to-date: don't set the switch if the
		 * tree has changed since we last opened cursors: that can lead
		 * to switching multiple times when only one switch is
		 * required, creating very small chunks.
		 */
		WT_RET(__wt_lsm_tree_readlock(session, lsm_tree));
		if (lsm_tree->nchunks == 0 ||
		    (clsm->dsk_gen == lsm_tree->dsk_gen &&
		    !F_ISSET(lsm_tree, WT_LSM_TREE_NEED_SWITCH))) {
			F_SET(lsm_tree, WT_LSM_TREE_NEED_SWITCH);
			ret = __wt_lsm_manager_push_entry(
			    session, WT_LSM_WORK_SWITCH, 0, lsm_tree);
		}
		WT_TRET(__wt_lsm_tree_readunlock(session, lsm_tree));
	}

	return (ret);
}

/*
 * __wt_clsm_await_switch --
 *	Wait for a switch to have completed in the LSM tree
 */
int
__wt_clsm_await_switch(WT_CURSOR_LSM *clsm)
{
	WT_LSM_TREE *lsm_tree;
	WT_SESSION_IMPL *session;
	int waited;

	lsm_tree = clsm->lsm_tree;
	session = (WT_SESSION_IMPL *)clsm->iface.session;

	/*
	 * If there is no primary chunk, or a chunk has overflowed the hard
	 * limit, which either means a worker thread has fallen behind or there
	 * has just been a user-level checkpoint, wait until the tree changes.
	 *
	 * We used to switch chunks in the application thread here, but that is
	 * problematic because there is a transaction in progress and it could
	 * roll back, leaving the metadata inconsistent.
	 */
	for (waited = 0;
	    lsm_tree->nchunks == 0 ||
	    clsm->dsk_gen == lsm_tree->dsk_gen;
	    ++waited) {
		if (waited % WT_THOUSAND == 0)
			WT_RET(__wt_lsm_manager_push_entry(
			    session, WT_LSM_WORK_SWITCH, 0, lsm_tree));
		__wt_sleep(0, 10);
	}
	return (0);
}

/*
 * __clsm_enter_update --
 *	Make sure an LSM cursor is ready to perform an update.
 */
static int
__clsm_enter_update(WT_CURSOR_LSM *clsm)
{
	WT_CURSOR *primary;
	WT_LSM_CHUNK *primary_chunk;
	WT_LSM_TREE *lsm_tree;
	WT_SESSION_IMPL *session;
	bool hard_limit, have_primary, ovfl;

	lsm_tree = clsm->lsm_tree;
	ovfl = false;
	session = (WT_SESSION_IMPL *)clsm->iface.session;

	if (clsm->nchunks == 0) {
		primary = NULL;
		have_primary = false;
	} else {
		primary = clsm->cursors[clsm->nchunks - 1];
		primary_chunk = clsm->primary_chunk;
		WT_ASSERT(session, F_ISSET(&session->txn, WT_TXN_HAS_ID));
		have_primary = (primary != NULL && primary_chunk != NULL &&
		    (primary_chunk->switch_txn == WT_TXN_NONE ||
		    WT_TXNID_LT(session->txn.id, primary_chunk->switch_txn)));
	}

	/*
	 * In LSM there are multiple btrees active at one time. The tree
	 * switch code needs to use btree API methods, and it wants to
	 * operate on the btree for the primary chunk. Set that up now.
	 *
	 * If the primary chunk has grown too large, set a flag so the worker
	 * thread will switch when it gets a chance to avoid introducing high
	 * latency into application threads.  Don't do this indefinitely: if a
	 * chunk grows twice as large as the configured size, block until it
	 * can be switched.
	 */
	hard_limit = F_ISSET(lsm_tree, WT_LSM_TREE_NEED_SWITCH);

	if (have_primary) {
		WT_ENTER_PAGE_INDEX(session);
		WT_WITH_BTREE(session, ((WT_CURSOR_BTREE *)primary)->btree,
		    ovfl = __wt_btree_lsm_over_size(session, hard_limit ?
		    2 * lsm_tree->chunk_size : lsm_tree->chunk_size));
		WT_LEAVE_PAGE_INDEX(session);

		/* If there was no overflow, we're done. */
		if (!ovfl)
			return (0);
	}

	/* Request a switch. */
	WT_RET(__wt_clsm_request_switch(clsm));

	/* If we only overflowed the soft limit, we're done. */
	if (have_primary && !hard_limit)
		return (0);

	WT_RET(__wt_clsm_await_switch(clsm));

	return (0);
}

/*
 * __clsm_enter --
 *	Start an operation on an LSM cursor, update if the tree has changed.
 */
static inline int
__clsm_enter(WT_CURSOR_LSM *clsm, bool reset, bool update)
{
	WT_DECL_RET;
	WT_LSM_TREE *lsm_tree;
	WT_SESSION_IMPL *session;
	WT_TXN *txn;
	uint64_t *switch_txnp;
	uint64_t snap_min;

	lsm_tree = clsm->lsm_tree;
	session = (WT_SESSION_IMPL *)clsm->iface.session;
	txn = &session->txn;

	/* Merge cursors never update. */
	if (F_ISSET(clsm, WT_CLSM_MERGE))
		return (0);

	if (reset) {
		WT_ASSERT(session, !F_ISSET(&clsm->iface,
		   WT_CURSTD_KEY_INT | WT_CURSTD_VALUE_INT));
		WT_RET(__clsm_reset_cursors(clsm, NULL));
	}

	for (;;) {
		/*
		 * If the cursor looks up-to-date, check if the cache is full.
		 * In case this call blocks, the check will be repeated before
		 * proceeding.
		 */
		if (clsm->dsk_gen != lsm_tree->dsk_gen &&
		    lsm_tree->nchunks != 0)
			goto open;

		if (clsm->dsk_gen != lsm_tree->dsk_gen &&
		    lsm_tree->nchunks != 0)
			goto open;

		/* Update the maximum transaction ID in the primary chunk. */
		if (update) {
			/*
			 * Ensure that there is a transaction snapshot active.
			 */
			WT_RET(__wt_txn_autocommit_check(session));
			WT_RET(__wt_txn_id_check(session));

			WT_RET(__clsm_enter_update(clsm));
			if (clsm->dsk_gen != clsm->lsm_tree->dsk_gen)
				goto open;

			if (txn->isolation == WT_ISO_SNAPSHOT)
				__wt_txn_cursor_op(session);

			/*
			 * Figure out how many updates are required for
			 * snapshot isolation.
			 *
			 * This is not a normal visibility check on the maximum
			 * transaction ID in each chunk: any transaction ID
			 * that overlaps with our snapshot is a potential
			 * conflict.
			 */
			clsm->nupdates = 1;
			if (txn->isolation == WT_ISO_SNAPSHOT &&
			    F_ISSET(clsm, WT_CLSM_OPEN_SNAPSHOT)) {
				WT_ASSERT(session,
				    F_ISSET(txn, WT_TXN_HAS_SNAPSHOT));
				snap_min = txn->snap_min;
				for (switch_txnp =
				    &clsm->switch_txn[clsm->nchunks - 2];
				    clsm->nupdates < clsm->nchunks;
				    clsm->nupdates++, switch_txnp--) {
					if (WT_TXNID_LT(*switch_txnp, snap_min))
						break;
					WT_ASSERT(session,
					    !__wt_txn_visible_all(
					    session, *switch_txnp));
				}
			}
		}

		/*
		 * Stop when we are up-to-date, as long as this is:
		 *   - a snapshot isolation update and the cursor is set up for
		 *     that;
		 *   - an update operation with a primary chunk, or
		 *   - a read operation and the cursor is open for reading.
		 */
		if ((!update ||
		    txn->isolation != WT_ISO_SNAPSHOT ||
		    F_ISSET(clsm, WT_CLSM_OPEN_SNAPSHOT)) &&
		    ((update && clsm->primary_chunk != NULL) ||
		    (!update && F_ISSET(clsm, WT_CLSM_OPEN_READ))))
			break;

open:		WT_WITH_SCHEMA_LOCK(session,
		    ret = __clsm_open_cursors(clsm, update, 0, 0));
		WT_RET(ret);
	}

	if (!F_ISSET(clsm, WT_CLSM_ACTIVE)) {
		WT_RET(__cursor_enter(session));
		F_SET(clsm, WT_CLSM_ACTIVE);
	}

	return (0);
}

/*
 * __clsm_leave --
 *	Finish an operation on an LSM cursor.
 */
static void
__clsm_leave(WT_CURSOR_LSM *clsm)
{
	WT_SESSION_IMPL *session;

	session = (WT_SESSION_IMPL *)clsm->iface.session;

	if (F_ISSET(clsm, WT_CLSM_ACTIVE)) {
		__cursor_leave(session);
		F_CLR(clsm, WT_CLSM_ACTIVE);
	}
}

/*
 * We need a tombstone to mark deleted records, and we use the special
 * value below for that purpose.  We use two 0x14 (Device Control 4) bytes to
 * minimize the likelihood of colliding with an application-chosen encoding
 * byte, if the application uses two leading DC4 byte for some reason, we'll do
 * a wasted data copy each time a new value is inserted into the object.
 */
static const WT_ITEM __tombstone = { "\x14\x14", 2, 0, NULL, 0 };

/*
 * __clsm_deleted --
 *	Check whether the current value is a tombstone.
 */
static inline bool
__clsm_deleted(WT_CURSOR_LSM *clsm, const WT_ITEM *item)
{
	return (!F_ISSET(clsm, WT_CLSM_MINOR_MERGE) &&
	    item->size == __tombstone.size &&
	    memcmp(item->data, __tombstone.data, __tombstone.size) == 0);
}

/*
 * __clsm_deleted_encode --
 *	Encode values that are in the encoded name space.
 */
static inline int
__clsm_deleted_encode(WT_SESSION_IMPL *session,
    const WT_ITEM *value, WT_ITEM *final_value, WT_ITEM **tmpp)
{
	WT_ITEM *tmp;

	/*
	 * If value requires encoding, get a scratch buffer of the right size
	 * and create a copy of the data with the first byte of the tombstone
	 * appended.
	 */
	if (value->size >= __tombstone.size &&
	    memcmp(value->data, __tombstone.data, __tombstone.size) == 0) {
		WT_RET(__wt_scr_alloc(session, value->size + 1, tmpp));
		tmp = *tmpp;

		memcpy(tmp->mem, value->data, value->size);
		memcpy((uint8_t *)tmp->mem + value->size, __tombstone.data, 1);
		final_value->data = tmp->mem;
		final_value->size = value->size + 1;
	} else {
		final_value->data = value->data;
		final_value->size = value->size;
	}

	return (0);
}

/*
 * __clsm_deleted_decode --
 *	Decode values that start with the tombstone.
 */
static inline void
__clsm_deleted_decode(WT_CURSOR_LSM *clsm, WT_ITEM *value)
{
	/*
	 * Take care with this check: when an LSM cursor is used for a merge,
	 * and/or to create a Bloom filter, it is valid to return the tombstone
	 * value.
	 */
	if (!F_ISSET(clsm, WT_CLSM_MERGE) &&
	    value->size > __tombstone.size &&
	    memcmp(value->data, __tombstone.data, __tombstone.size) == 0)
		--value->size;
}

/*
 * __clsm_close_cursors --
 *	Close any btree cursors that are not needed.
 */
static int
__clsm_close_cursors(WT_CURSOR_LSM *clsm, u_int start, u_int end)
{
	WT_BLOOM *bloom;
	WT_CURSOR *c;
	u_int i;

	if (clsm->cursors == NULL || clsm->nchunks == 0)
		return (0);

	/*
	 * Walk the cursors, closing any we don't need.  Note that the exit
	 * condition here is special, don't use WT_FORALL_CURSORS, and be
	 * careful with unsigned integer wrapping.
	 */
	for (i = start; i < end; i++) {
		if ((c = (clsm)->cursors[i]) != NULL) {
			clsm->cursors[i] = NULL;
			WT_RET(c->close(c));
		}
		if ((bloom = clsm->blooms[i]) != NULL) {
			clsm->blooms[i] = NULL;
			WT_RET(__wt_bloom_close(bloom));
		}
	}

	return (0);
}

/*
 * __clsm_open_cursors --
 *	Open cursors for the current set of files.
 */
static int
__clsm_open_cursors(
    WT_CURSOR_LSM *clsm, bool update, u_int start_chunk, uint32_t start_id)
{
	WT_BTREE *btree;
	WT_CURSOR *c, **cp, *primary;
	WT_DECL_RET;
	WT_LSM_CHUNK *chunk;
	WT_LSM_TREE *lsm_tree;
	WT_SESSION_IMPL *session;
	WT_TXN *txn;
	const char *checkpoint, *ckpt_cfg[3];
	uint64_t saved_gen;
	u_int i, nchunks, ngood, nupdates;
	u_int close_range_end, close_range_start;
	bool locked;

	c = &clsm->iface;
	session = (WT_SESSION_IMPL *)c->session;
	txn = &session->txn;
	chunk = NULL;
	locked = false;
	lsm_tree = clsm->lsm_tree;

	/*
	 * Ensure that any snapshot update has cursors on the right set of
	 * chunks to guarantee visibility is correct.
	 */
	if (update && txn->isolation == WT_ISO_SNAPSHOT)
		F_SET(clsm, WT_CLSM_OPEN_SNAPSHOT);

	/*
	 * Query operations need a full set of cursors. Overwrite cursors
	 * do queries in service of updates.
	 */
	if (!update || !F_ISSET(c, WT_CURSTD_OVERWRITE))
		F_SET(clsm, WT_CLSM_OPEN_READ);

	if (lsm_tree->nchunks == 0)
		return (0);

	ckpt_cfg[0] = WT_CONFIG_BASE(session, WT_SESSION_open_cursor);
	ckpt_cfg[1] = "checkpoint=" WT_CHECKPOINT ",raw";
	ckpt_cfg[2] = NULL;

	/*
	 * If the key is pointing to memory that is pinned by a chunk
	 * cursor, take a copy before closing cursors.
	 */
	if (F_ISSET(c, WT_CURSTD_KEY_INT))
		WT_CURSOR_NEEDKEY(c);

	F_CLR(clsm, WT_CLSM_ITERATE_NEXT | WT_CLSM_ITERATE_PREV);

	WT_RET(__wt_lsm_tree_readlock(session, lsm_tree));
	locked = true;

	/* Merge cursors have already figured out how many chunks they need. */
retry:	if (F_ISSET(clsm, WT_CLSM_MERGE)) {
		nchunks = clsm->nchunks;
		ngood = 0;

		/*
		 * We may have raced with another merge completing.  Check that
		 * we're starting at the right offset in the chunk array.
		 */
		if (start_chunk >= lsm_tree->nchunks ||
		    lsm_tree->chunk[start_chunk]->id != start_id) {
			for (start_chunk = 0;
			    start_chunk < lsm_tree->nchunks;
			    start_chunk++) {
				chunk = lsm_tree->chunk[start_chunk];
				if (chunk->id == start_id)
					break;
			}
			/* We have to find the start chunk: merge locked it. */
			WT_ASSERT(session, start_chunk < lsm_tree->nchunks);
		}

		WT_ASSERT(session, start_chunk + nchunks <= lsm_tree->nchunks);
	} else {
		nchunks = lsm_tree->nchunks;

		/*
		 * If we are only opening the cursor for updates, only open the
		 * primary chunk, plus any other chunks that might be required
		 * to detect snapshot isolation conflicts.
		 */
		if (F_ISSET(clsm, WT_CLSM_OPEN_SNAPSHOT))
			WT_ERR(__wt_realloc_def(session,
			    &clsm->txnid_alloc, nchunks,
			    &clsm->switch_txn));
		if (F_ISSET(clsm, WT_CLSM_OPEN_READ))
			ngood = nupdates = 0;
		else if (F_ISSET(clsm, WT_CLSM_OPEN_SNAPSHOT)) {
			/*
			 * Keep going until all updates in the next
			 * chunk are globally visible.  Copy the maximum
			 * transaction IDs into the cursor as we go.
			 */
			for (ngood = nchunks - 1, nupdates = 1;
			    ngood > 0;
			    ngood--, nupdates++) {
				chunk = lsm_tree->chunk[ngood - 1];
				clsm->switch_txn[ngood - 1] = chunk->switch_txn;
				if (__wt_txn_visible_all(
				    session, chunk->switch_txn))
					break;
			}
		} else {
			nupdates = 1;
			ngood = nchunks - 1;
		}

		/* Check how many cursors are already open. */
		for (cp = clsm->cursors + ngood;
		    ngood < clsm->nchunks && ngood < nchunks;
		    cp++, ngood++) {
			chunk = lsm_tree->chunk[ngood];

			/* If the cursor isn't open yet, we're done. */
			if (*cp == NULL)
				break;

			/* Easy case: the URIs don't match. */
			if (strcmp((*cp)->uri, chunk->uri) != 0)
				break;

			/* Make sure the checkpoint config matches. */
			checkpoint = ((WT_CURSOR_BTREE *)*cp)->
			    btree->dhandle->checkpoint;
			if (checkpoint == NULL &&
			    F_ISSET(chunk, WT_LSM_CHUNK_ONDISK) &&
			    !chunk->empty)
				break;

			/* Make sure the Bloom config matches. */
			if (clsm->blooms[ngood] == NULL &&
			    F_ISSET(chunk, WT_LSM_CHUNK_BLOOM))
				break;
		}

		/* Spurious generation bump? */
		if (ngood == clsm->nchunks && clsm->nchunks == nchunks) {
			clsm->dsk_gen = lsm_tree->dsk_gen;
			goto err;
		}

		/*
		 * Close any cursors we no longer need.
		 *
		 * Drop the LSM tree lock while we do this: if the cache is
		 * full, we may block while closing a cursor.  Save the
		 * generation number and retry if it has changed under us.
		 */
		if (clsm->cursors != NULL && ngood < clsm->nchunks) {
			close_range_start = ngood;
			close_range_end = clsm->nchunks;
		} else if (!F_ISSET(clsm, WT_CLSM_OPEN_READ) && nupdates > 0 ) {
			close_range_start = 0;
			close_range_end = WT_MIN(nchunks, clsm->nchunks);
			if (close_range_end > nupdates)
				close_range_end -= nupdates;
			else
				close_range_end = 0;
			WT_ASSERT(session, ngood >= close_range_end);
		} else {
			close_range_end = 0;
			close_range_start = 0;
		}
		if (close_range_end > close_range_start) {
			saved_gen = lsm_tree->dsk_gen;
			locked = false;
			WT_ERR(__wt_lsm_tree_readunlock(session, lsm_tree));
			WT_ERR(__clsm_close_cursors(
			    clsm, close_range_start, close_range_end));
			WT_ERR(__wt_lsm_tree_readlock(session, lsm_tree));
			locked = true;
			if (lsm_tree->dsk_gen != saved_gen)
				goto retry;
		}

		/* Detach from our old primary. */
		clsm->primary_chunk = NULL;
		clsm->current = NULL;
	}

	WT_ERR(__wt_realloc_def(session,
	    &clsm->bloom_alloc, nchunks, &clsm->blooms));
	WT_ERR(__wt_realloc_def(session,
	    &clsm->cursor_alloc, nchunks, &clsm->cursors));

	clsm->nchunks = nchunks;

	/* Open the cursors for chunks that have changed. */
	for (i = ngood, cp = clsm->cursors + i; i != nchunks; i++, cp++) {
		chunk = lsm_tree->chunk[i + start_chunk];
		/* Copy the maximum transaction ID. */
		if (F_ISSET(clsm, WT_CLSM_OPEN_SNAPSHOT))
			clsm->switch_txn[i] = chunk->switch_txn;

		/*
		 * Read from the checkpoint if the file has been written.
		 * Once all cursors switch, the in-memory tree can be evicted.
		 */
		WT_ASSERT(session, *cp == NULL);
		ret = __wt_open_cursor(session, chunk->uri, c,
		    (F_ISSET(chunk, WT_LSM_CHUNK_ONDISK) && !chunk->empty) ?
			ckpt_cfg : NULL, cp);

		/*
		 * XXX kludge: we may have an empty chunk where no checkpoint
		 * was written.  If so, try to open the ordinary handle on that
		 * chunk instead.
		 */
		if (ret == WT_NOTFOUND && F_ISSET(chunk, WT_LSM_CHUNK_ONDISK)) {
			ret = __wt_open_cursor(
			    session, chunk->uri, c, NULL, cp);
			if (ret == 0)
				chunk->empty = 1;
		}
		WT_ERR(ret);

		/*
		 * Setup all cursors other than the primary to only do conflict
		 * checks on insert operations. This allows us to execute
		 * inserts on non-primary chunks as a way of checking for
		 * write conflicts with concurrent updates.
		 */
		if (i != nchunks - 1)
			(*cp)->insert = __wt_curfile_update_check;

		if (!F_ISSET(clsm, WT_CLSM_MERGE) &&
		    F_ISSET(chunk, WT_LSM_CHUNK_BLOOM))
			WT_ERR(__wt_bloom_open(session, chunk->bloom_uri,
			    lsm_tree->bloom_bit_count,
			    lsm_tree->bloom_hash_count,
			    c, &clsm->blooms[i]));

		/* Child cursors always use overwrite and raw mode. */
		F_SET(*cp, WT_CURSTD_OVERWRITE | WT_CURSTD_RAW);
	}

	/* The last chunk is our new primary. */
	if (chunk != NULL &&
	    !F_ISSET(chunk, WT_LSM_CHUNK_ONDISK) &&
	    chunk->switch_txn == WT_TXN_NONE) {
		clsm->primary_chunk = chunk;
		primary = clsm->cursors[clsm->nchunks - 1];
		/*
		 * Disable eviction for the in-memory chunk.  Also clear the
		 * bulk load flag here, otherwise eviction will be enabled by
		 * the first update.
		 */
		btree = ((WT_CURSOR_BTREE *)(primary))->btree;
		if (btree->bulk_load_ok) {
			btree->bulk_load_ok = false;
			WT_WITH_BTREE(session, btree,
			    __wt_btree_evictable(session, false));
		}
	}

	clsm->dsk_gen = lsm_tree->dsk_gen;

err:	
#ifdef HAVE_DIAGNOSTIC
	/* Check that all cursors are open as expected. */
	if (ret == 0 && F_ISSET(clsm, WT_CLSM_OPEN_READ)) {
		for (i = 0, cp = clsm->cursors; i != clsm->nchunks; cp++, i++) {
			chunk = lsm_tree->chunk[i + start_chunk];

			/* Make sure the cursor is open. */
			WT_ASSERT(session, *cp != NULL);

			/* Easy case: the URIs should match. */
			WT_ASSERT(session, strcmp((*cp)->uri, chunk->uri) == 0);

			/* Make sure the checkpoint config matches. */
			checkpoint = ((WT_CURSOR_BTREE *)*cp)->
			    btree->dhandle->checkpoint;
			WT_ASSERT(session,
			    (F_ISSET(chunk, WT_LSM_CHUNK_ONDISK) &&
			    !chunk->empty) ?
			    checkpoint != NULL : checkpoint == NULL);

			/* Make sure the Bloom config matches. */
			WT_ASSERT(session,
			    (F_ISSET(chunk, WT_LSM_CHUNK_BLOOM) &&
			    !F_ISSET(clsm, WT_CLSM_MERGE)) ?
			    clsm->blooms[i] != NULL : clsm->blooms[i] == NULL);
		}
	}
#endif
	if (locked)
		WT_TRET(__wt_lsm_tree_readunlock(session, lsm_tree));
	return (ret);
}

/*
 * __wt_clsm_init_merge --
 *	Initialize an LSM cursor for a merge.
 */
int
__wt_clsm_init_merge(
    WT_CURSOR *cursor, u_int start_chunk, uint32_t start_id, u_int nchunks)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	clsm = (WT_CURSOR_LSM *)cursor;
	session = (WT_SESSION_IMPL *)cursor->session;

	F_SET(clsm, WT_CLSM_MERGE);
	if (start_chunk != 0)
		F_SET(clsm, WT_CLSM_MINOR_MERGE);
	clsm->nchunks = nchunks;

	WT_WITH_SCHEMA_LOCK(session,
	    ret = __clsm_open_cursors(clsm, false, start_chunk, start_id));
	return (ret);
}

/*
 * __clsm_get_current --
 *	Find the smallest / largest of the cursors and copy its key/value.
 */
static int
__clsm_get_current(WT_SESSION_IMPL *session,
    WT_CURSOR_LSM *clsm, bool smallest, bool *deletedp)
{
	WT_CURSOR *c, *current;
	int cmp;
	u_int i;
	bool multiple;

	current = NULL;
	multiple = false;

	WT_FORALL_CURSORS(clsm, c, i) {
		if (!F_ISSET(c, WT_CURSTD_KEY_INT))
			continue;
		if (current == NULL) {
			current = c;
			continue;
		}
		WT_RET(WT_LSM_CURCMP(session, clsm->lsm_tree, c, current, cmp));
		if (smallest ? cmp < 0 : cmp > 0) {
			current = c;
			multiple = false;
		} else if (cmp == 0)
			multiple = true;
	}

	c = &clsm->iface;
	if ((clsm->current = current) == NULL) {
		F_CLR(c, WT_CURSTD_KEY_SET | WT_CURSTD_VALUE_SET);
		return (WT_NOTFOUND);
	}

	if (multiple)
		F_SET(clsm, WT_CLSM_MULTIPLE);
	else
		F_CLR(clsm, WT_CLSM_MULTIPLE);

	WT_RET(current->get_key(current, &c->key));
	WT_RET(current->get_value(current, &c->value));

	F_CLR(c, WT_CURSTD_KEY_SET | WT_CURSTD_VALUE_SET);
	if ((*deletedp = __clsm_deleted(clsm, &c->value)) == false)
		F_SET(c, WT_CURSTD_KEY_INT | WT_CURSTD_VALUE_INT);

	return (0);
}

/*
 * __clsm_compare --
 *	WT_CURSOR->compare implementation for the LSM cursor type.
 */
static int
__clsm_compare(WT_CURSOR *a, WT_CURSOR *b, int *cmpp)
{
	WT_CURSOR_LSM *alsm;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	/* There's no need to sync with the LSM tree, avoid WT_LSM_ENTER. */
	alsm = (WT_CURSOR_LSM *)a;
	CURSOR_API_CALL(a, session, compare, NULL);

	/*
	 * Confirm both cursors refer to the same source and have keys, then
	 * compare the keys.
	 */
	if (strcmp(a->uri, b->uri) != 0)
		WT_ERR_MSG(session, EINVAL,
		    "comparison method cursors must reference the same object");

	WT_CURSOR_NEEDKEY(a);
	WT_CURSOR_NEEDKEY(b);

	WT_ERR(__wt_compare(
	    session, alsm->lsm_tree->collator, &a->key, &b->key, cmpp));

err:	API_END_RET(session, ret);
}

/*
 * __clsm_next --
 *	WT_CURSOR->next method for the LSM cursor type.
 */
static int
__clsm_next(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_CURSOR *c;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;
	u_int i;
	int cmp;
	bool check, deleted;

	clsm = (WT_CURSOR_LSM *)cursor;

	CURSOR_API_CALL(cursor, session, next, NULL);
	WT_CURSOR_NOVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, false, false));

	/* If we aren't positioned for a forward scan, get started. */
	if (clsm->current == NULL || !F_ISSET(clsm, WT_CLSM_ITERATE_NEXT)) {
		F_CLR(clsm, WT_CLSM_MULTIPLE);
		WT_FORALL_CURSORS(clsm, c, i) {
			if (!F_ISSET(cursor, WT_CURSTD_KEY_SET)) {
				WT_ERR(c->reset(c));
				ret = c->next(c);
			} else if (c != clsm->current) {
				c->set_key(c, &cursor->key);
				if ((ret = c->search_near(c, &cmp)) == 0) {
					if (cmp < 0)
						ret = c->next(c);
					else if (cmp == 0) {
						if (clsm->current == NULL)
							clsm->current = c;
						else
							F_SET(clsm,
							    WT_CLSM_MULTIPLE);
					}
				} else
					F_CLR(c, WT_CURSTD_KEY_SET);
			}
			WT_ERR_NOTFOUND_OK(ret);
		}
		F_SET(clsm, WT_CLSM_ITERATE_NEXT);
		F_CLR(clsm, WT_CLSM_ITERATE_PREV);

		/* We just positioned *at* the key, now move. */
		if (clsm->current != NULL)
			goto retry;
	} else {
retry:		/*
		 * If there are multiple cursors on that key, move them
		 * forward.
		 */
		if (F_ISSET(clsm, WT_CLSM_MULTIPLE)) {
			check = false;
			WT_FORALL_CURSORS(clsm, c, i) {
				if (!F_ISSET(c, WT_CURSTD_KEY_INT))
					continue;
				if (check) {
					WT_ERR(WT_LSM_CURCMP(session,
					    clsm->lsm_tree, c, clsm->current,
					    cmp));
					if (cmp == 0)
						WT_ERR_NOTFOUND_OK(c->next(c));
				}
				if (c == clsm->current)
					check = true;
			}
		}

		/* Move the smallest cursor forward. */
		c = clsm->current;
		WT_ERR_NOTFOUND_OK(c->next(c));
	}

	/* Find the cursor(s) with the smallest key. */
	if ((ret = __clsm_get_current(session, clsm, true, &deleted)) == 0 &&
	    deleted)
		goto retry;

err:	__clsm_leave(clsm);
	API_END(session, ret);
	if (ret == 0)
		__clsm_deleted_decode(clsm, &cursor->value);
	return (ret);
}

/*
 * __clsm_prev --
 *	WT_CURSOR->prev method for the LSM cursor type.
 */
static int
__clsm_prev(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_CURSOR *c;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;
	u_int i;
	int cmp;
	bool check, deleted;

	clsm = (WT_CURSOR_LSM *)cursor;

	CURSOR_API_CALL(cursor, session, prev, NULL);
	WT_CURSOR_NOVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, false, false));

	/* If we aren't positioned for a reverse scan, get started. */
	if (clsm->current == NULL || !F_ISSET(clsm, WT_CLSM_ITERATE_PREV)) {
		F_CLR(clsm, WT_CLSM_MULTIPLE);
		WT_FORALL_CURSORS(clsm, c, i) {
			if (!F_ISSET(cursor, WT_CURSTD_KEY_SET)) {
				WT_ERR(c->reset(c));
				ret = c->prev(c);
			} else if (c != clsm->current) {
				c->set_key(c, &cursor->key);
				if ((ret = c->search_near(c, &cmp)) == 0) {
					if (cmp > 0)
						ret = c->prev(c);
					else if (cmp == 0) {
						if (clsm->current == NULL)
							clsm->current = c;
						else
							F_SET(clsm,
							    WT_CLSM_MULTIPLE);
					}
				}
			}
			WT_ERR_NOTFOUND_OK(ret);
		}
		F_SET(clsm, WT_CLSM_ITERATE_PREV);
		F_CLR(clsm, WT_CLSM_ITERATE_NEXT);

		/* We just positioned *at* the key, now move. */
		if (clsm->current != NULL)
			goto retry;
	} else {
retry:		/*
		 * If there are multiple cursors on that key, move them
		 * backwards.
		 */
		if (F_ISSET(clsm, WT_CLSM_MULTIPLE)) {
			check = false;
			WT_FORALL_CURSORS(clsm, c, i) {
				if (!F_ISSET(c, WT_CURSTD_KEY_INT))
					continue;
				if (check) {
					WT_ERR(WT_LSM_CURCMP(session,
					    clsm->lsm_tree, c, clsm->current,
					    cmp));
					if (cmp == 0)
						WT_ERR_NOTFOUND_OK(c->prev(c));
				}
				if (c == clsm->current)
					check = true;
			}
		}

		/* Move the smallest cursor backwards. */
		c = clsm->current;
		WT_ERR_NOTFOUND_OK(c->prev(c));
	}

	/* Find the cursor(s) with the largest key. */
	if ((ret = __clsm_get_current(session, clsm, false, &deleted)) == 0 &&
	    deleted)
		goto retry;

err:	__clsm_leave(clsm);
	API_END(session, ret);
	if (ret == 0)
		__clsm_deleted_decode(clsm, &cursor->value);
	return (ret);
}

/*
 * __clsm_reset_cursors --
 *	Reset any positioned chunk cursors.
 *
 *	If the skip parameter is non-NULL, that cursor is about to be used, so
 *	there is no need to reset it.
 */
static int
__clsm_reset_cursors(WT_CURSOR_LSM *clsm, WT_CURSOR *skip)
{
	WT_CURSOR *c;
	WT_DECL_RET;
	u_int i;

	/* Fast path if the cursor is not positioned. */
	if ((clsm->current == NULL || clsm->current == skip) &&
	    !F_ISSET(clsm, WT_CLSM_ITERATE_NEXT | WT_CLSM_ITERATE_PREV))
		return (0);

	WT_FORALL_CURSORS(clsm, c, i) {
		if (c == skip)
			continue;
		if (F_ISSET(c, WT_CURSTD_KEY_INT))
			WT_TRET(c->reset(c));
	}

	clsm->current = NULL;
	F_CLR(clsm, WT_CLSM_ITERATE_NEXT | WT_CLSM_ITERATE_PREV);

	return (ret);
}

/*
 * __clsm_reset --
 *	WT_CURSOR->reset method for the LSM cursor type.
 */
static int
__clsm_reset(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	/*
	 * Don't use the normal __clsm_enter path: that is wasted work when all
	 * we want to do is give up our position.
	 */
	clsm = (WT_CURSOR_LSM *)cursor;
	CURSOR_API_CALL(cursor, session, reset, NULL);
	F_CLR(cursor, WT_CURSTD_KEY_SET | WT_CURSTD_VALUE_SET);

	WT_TRET(__clsm_reset_cursors(clsm, NULL));

	/* In case we were left positioned, clear that. */
	__clsm_leave(clsm);

err:	API_END_RET(session, ret);
}

/*
 * __clsm_lookup --
 *	Position an LSM cursor.
 */
static int
__clsm_lookup(WT_CURSOR_LSM *clsm, WT_ITEM *value)
{
	WT_BLOOM *bloom;
	WT_BLOOM_HASH bhash;
	WT_CURSOR *c, *cursor;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;
	u_int i;
	bool have_hash;

	c = NULL;
	cursor = &clsm->iface;
	have_hash = false;
	session = (WT_SESSION_IMPL *)cursor->session;

	WT_FORALL_CURSORS(clsm, c, i) {
		/* If there is a Bloom filter, see if we can skip the read. */
		bloom = NULL;
		if ((bloom = clsm->blooms[i]) != NULL) {
			if (!have_hash) {
				WT_ERR(__wt_bloom_hash(
				    bloom, &cursor->key, &bhash));
				have_hash = true;
			}

			ret = __wt_bloom_hash_get(bloom, &bhash);
			if (ret == WT_NOTFOUND) {
				WT_LSM_TREE_STAT_INCR(
				    session, clsm->lsm_tree->bloom_miss);
				continue;
			} else if (ret == 0)
				WT_LSM_TREE_STAT_INCR(
				    session, clsm->lsm_tree->bloom_hit);
			WT_ERR(ret);
		}
		c->set_key(c, &cursor->key);
		if ((ret = c->search(c)) == 0) {
			WT_ERR(c->get_key(c, &cursor->key));
			WT_ERR(c->get_value(c, value));
			if (__clsm_deleted(clsm, value))
				ret = WT_NOTFOUND;
			goto done;
		}
		WT_ERR_NOTFOUND_OK(ret);
		F_CLR(c, WT_CURSTD_KEY_SET);
		/* Update stats: the active chunk can't have a bloom filter. */
		if (bloom != NULL)
			WT_LSM_TREE_STAT_INCR(session,
			    clsm->lsm_tree->bloom_false_positive);
		else if (clsm->primary_chunk == NULL || i != clsm->nchunks)
			WT_LSM_TREE_STAT_INCR(session,
			    clsm->lsm_tree->lsm_lookup_no_bloom);
	}
	WT_ERR(WT_NOTFOUND);

done:
err:	F_CLR(cursor, WT_CURSTD_KEY_SET | WT_CURSTD_VALUE_SET);
	if (ret == 0) {
		clsm->current = c;
		F_SET(cursor, WT_CURSTD_KEY_INT);
		if (value == &cursor->value)
			F_SET(cursor, WT_CURSTD_VALUE_INT);
	} else if (c != NULL)
		WT_TRET(c->reset(c));

	return (ret);
}

/*
 * __clsm_search --
 *	WT_CURSOR->search method for the LSM cursor type.
 */
static int
__clsm_search(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	clsm = (WT_CURSOR_LSM *)cursor;

	CURSOR_API_CALL(cursor, session, search, NULL);
	WT_CURSOR_NEEDKEY(cursor);
	WT_CURSOR_NOVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, true, false));

	ret = __clsm_lookup(clsm, &cursor->value);

err:	__clsm_leave(clsm);
	API_END(session, ret);
	if (ret == 0)
		__clsm_deleted_decode(clsm, &cursor->value);
	return (ret);
}

/*
 * __clsm_search_near --
 *	WT_CURSOR->search_near method for the LSM cursor type.
 */
static int
__clsm_search_near(WT_CURSOR *cursor, int *exactp)
{
	WT_CURSOR *c, *closest;
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;
	u_int i;
	int cmp, exact;
	bool deleted;

	closest = NULL;
	clsm = (WT_CURSOR_LSM *)cursor;
	exact = 0;
	deleted = false;

	CURSOR_API_CALL(cursor, session, search_near, NULL);
	WT_CURSOR_NEEDKEY(cursor);
	WT_CURSOR_NOVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, true, false));
	F_CLR(clsm, WT_CLSM_ITERATE_NEXT | WT_CLSM_ITERATE_PREV);

	/*
	 * search_near is somewhat fiddly: we can't just use a nearby key from
	 * the in-memory chunk because there could be a closer key on disk.
	 *
	 * As we search down the chunks, we stop as soon as we find an exact
	 * match.  Otherwise, we maintain the smallest cursor larger than the
	 * search key and the largest cursor smaller than the search key.  At
	 * the end, we prefer the larger cursor, but if no record is larger,
	 * position on the last record in the tree.
	 */
	WT_FORALL_CURSORS(clsm, c, i) {
		c->set_key(c, &cursor->key);
		if ((ret = c->search_near(c, &cmp)) == WT_NOTFOUND) {
			ret = 0;
			continue;
		} else if (ret != 0)
			goto err;

		/* Do we have an exact match? */
		if (cmp == 0) {
			closest = c;
			exact = 1;
			break;
		}

		/*
		 * Prefer larger cursors.  There are two reasons: (1) we expect
		 * prefix searches to be a common case (as in our own indices);
		 * and (2) we need a way to unambiguously know we have the
		 * "closest" result.
		 */
		if (cmp < 0) {
			if ((ret = c->next(c)) == WT_NOTFOUND) {
				ret = 0;
				continue;
			} else if (ret != 0)
				goto err;
		}

		/*
		 * We are trying to find the smallest cursor greater than the
		 * search key.
		 */
		if (closest == NULL)
			closest = c;
		else {
			WT_ERR(WT_LSM_CURCMP(session,
			    clsm->lsm_tree, c, closest, cmp));
			if (cmp < 0)
				closest = c;
		}
	}

	/*
	 * At this point, we either have an exact match, or closest is the
	 * smallest cursor larger than the search key, or it is NULL if the
	 * search key is larger than any record in the tree.
	 */
	cmp = exact ? 0 : 1;

	/*
	 * If we land on a deleted item, try going forwards or backwards to
	 * find one that isn't deleted.  If the whole tree is empty, we'll
	 * end up with WT_NOTFOUND, as expected.
	 */
	if (closest == NULL)
		deleted = true;
	else {
		WT_ERR(closest->get_key(closest, &cursor->key));
		WT_ERR(closest->get_value(closest, &cursor->value));
		clsm->current = closest;
		closest = NULL;
		deleted = __clsm_deleted(clsm, &cursor->value);
		if (!deleted)
			__clsm_deleted_decode(clsm, &cursor->value);
		else  {
			/*
			 * We have a key pointing at memory that is
			 * pinned by the current chunk cursor.  In the
			 * unlikely event that we have to reopen cursors
			 * to move to the next record, make sure the cursor
			 * flags are set so a copy is made before the current
			 * chunk cursor releases its position.
			 */
			F_CLR(cursor, WT_CURSTD_KEY_SET);
			F_SET(cursor, WT_CURSTD_KEY_INT);
			if ((ret = cursor->next(cursor)) == 0) {
				cmp = 1;
				deleted = false;
			}
		}
		WT_ERR_NOTFOUND_OK(ret);
	}
	if (deleted) {
		clsm->current = NULL;
		WT_ERR(cursor->prev(cursor));
		cmp = -1;
	}
	*exactp = cmp;

err:	__clsm_leave(clsm);
	API_END(session, ret);
	if (closest != NULL)
		WT_TRET(closest->reset(closest));

	F_CLR(cursor, WT_CURSTD_KEY_SET | WT_CURSTD_VALUE_SET);
	if (ret == 0) {
		F_SET(cursor, WT_CURSTD_KEY_INT | WT_CURSTD_VALUE_INT);
	} else
		clsm->current = NULL;

	return (ret);
}

/*
 * __clsm_put --
 *	Put an entry into the in-memory tree, trigger a file switch if
 *	necessary.
 */
static inline int
__clsm_put(WT_SESSION_IMPL *session, WT_CURSOR_LSM *clsm,
    const WT_ITEM *key, const WT_ITEM *value, bool position)
{
	WT_CURSOR *c, *primary;
	WT_LSM_TREE *lsm_tree;
	u_int i, slot;

	lsm_tree = clsm->lsm_tree;

	WT_ASSERT(session,
	    F_ISSET(&session->txn, WT_TXN_HAS_ID) &&
	    clsm->primary_chunk != NULL &&
	    (clsm->primary_chunk->switch_txn == WT_TXN_NONE ||
	    WT_TXNID_LE(session->txn.id, clsm->primary_chunk->switch_txn)));

	/*
	 * Clear the existing cursor position.  Don't clear the primary cursor:
	 * we're about to use it anyway.
	 */
	primary = clsm->cursors[clsm->nchunks - 1];
	WT_RET(__clsm_reset_cursors(clsm, primary));

	/* If necessary, set the position for future scans. */
	if (position)
		clsm->current = primary;

	for (i = 0, slot = clsm->nchunks - 1; i < clsm->nupdates; i++, slot--) {
		/* Check if we need to keep updating old chunks. */
		if (i > 0 &&
		    __wt_txn_visible(session, clsm->switch_txn[slot])) {
			clsm->nupdates = i;
			break;
		}

		c = clsm->cursors[slot];
		c->set_key(c, key);
		c->set_value(c, value);
		WT_RET((position && i == 0) ? c->update(c) : c->insert(c));
	}

	/*
	 * Update the record count.  It is in a shared structure, but it's only
	 * approximate, so don't worry about protecting access.
	 *
	 * Throttle if necessary.  Every 100 update operations on each cursor,
	 * check if throttling is required.  Don't rely only on the shared
	 * counter because it can race, and because for some workloads, there
	 * may not be enough records per chunk to get effective throttling.
	 */
	if ((++clsm->primary_chunk->count % 100 == 0 ||
	    ++clsm->update_count >= 100) &&
	    lsm_tree->merge_throttle + lsm_tree->ckpt_throttle > 0) {
		clsm->update_count = 0;
		WT_LSM_TREE_STAT_INCRV(session,
		    lsm_tree->lsm_checkpoint_throttle, lsm_tree->ckpt_throttle);
		WT_STAT_FAST_CONN_INCRV(session,
		    lsm_checkpoint_throttle, lsm_tree->ckpt_throttle);
		WT_LSM_TREE_STAT_INCRV(session,
		    lsm_tree->lsm_merge_throttle, lsm_tree->merge_throttle);
		WT_STAT_FAST_CONN_INCRV(session,
		    lsm_merge_throttle, lsm_tree->merge_throttle);
		__wt_sleep(0,
		    lsm_tree->ckpt_throttle + lsm_tree->merge_throttle);
	}

	return (0);
}

/*
 * __clsm_insert --
 *	WT_CURSOR->insert method for the LSM cursor type.
 */
static int
__clsm_insert(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_ITEM(buf);
	WT_DECL_RET;
	WT_ITEM value;
	WT_SESSION_IMPL *session;

	clsm = (WT_CURSOR_LSM *)cursor;

	CURSOR_UPDATE_API_CALL(cursor, session, insert, NULL);
	WT_CURSOR_NEEDKEY(cursor);
	WT_CURSOR_NEEDVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, false, true));

	if (!F_ISSET(cursor, WT_CURSTD_OVERWRITE) &&
	    (ret = __clsm_lookup(clsm, &value)) != WT_NOTFOUND) {
		if (ret == 0)
			ret = WT_DUPLICATE_KEY;
		goto err;
	}

	WT_ERR(__clsm_deleted_encode(session, &cursor->value, &value, &buf));
	WT_ERR(__clsm_put(session, clsm, &cursor->key, &value, false));

	/*
	 * WT_CURSOR.insert doesn't leave the cursor positioned, and the
	 * application may want to free the memory used to configure the
	 * insert; don't read that memory again (matching the underlying
	 * file object cursor insert semantics).
	 */
	F_CLR(cursor, WT_CURSTD_KEY_SET | WT_CURSTD_VALUE_SET);

err:	__wt_scr_free(session, &buf);
	__clsm_leave(clsm);
	CURSOR_UPDATE_API_END(session, ret);
	return (ret);
}

/*
 * __clsm_update --
 *	WT_CURSOR->update method for the LSM cursor type.
 */
static int
__clsm_update(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_ITEM(buf);
	WT_DECL_RET;
	WT_ITEM value;
	WT_SESSION_IMPL *session;

	clsm = (WT_CURSOR_LSM *)cursor;

	CURSOR_UPDATE_API_CALL(cursor, session, update, NULL);
	WT_CURSOR_NEEDKEY(cursor);
	WT_CURSOR_NEEDVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, false, true));

	if (F_ISSET(cursor, WT_CURSTD_OVERWRITE) ||
	    (ret = __clsm_lookup(clsm, &value)) == 0) {
		WT_ERR(__clsm_deleted_encode(
		    session, &cursor->value, &value, &buf));
		ret = __clsm_put(session, clsm, &cursor->key, &value, true);
	}

err:	__wt_scr_free(session, &buf);
	__clsm_leave(clsm);
	CURSOR_UPDATE_API_END(session, ret);
	return (ret);
}

/*
 * __clsm_remove --
 *	WT_CURSOR->remove method for the LSM cursor type.
 */
static int
__clsm_remove(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_ITEM value;
	WT_SESSION_IMPL *session;

	clsm = (WT_CURSOR_LSM *)cursor;

	CURSOR_REMOVE_API_CALL(cursor, session, NULL);
	WT_CURSOR_NEEDKEY(cursor);
	WT_CURSOR_NOVALUE(cursor);
	WT_ERR(__clsm_enter(clsm, false, true));

	if (F_ISSET(cursor, WT_CURSTD_OVERWRITE) ||
	    (ret = __clsm_lookup(clsm, &value)) == 0)
		ret = __clsm_put(
		    session, clsm, &cursor->key, &__tombstone, true);

err:	__clsm_leave(clsm);
	CURSOR_UPDATE_API_END(session, ret);
	return (ret);
}

/*
 * __wt_clsm_close --
 *	WT_CURSOR->close method for the LSM cursor type.
 */
int
__wt_clsm_close(WT_CURSOR *cursor)
{
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_SESSION_IMPL *session;

	/*
	 * Don't use the normal __clsm_enter path: that is wasted work when
	 * closing, and the cursor may never have been used.
	 */
	clsm = (WT_CURSOR_LSM *)cursor;
	CURSOR_API_CALL(cursor, session, close, NULL);
	WT_TRET(__clsm_close_cursors(clsm, 0, clsm->nchunks));
	__wt_free(session, clsm->blooms);
	__wt_free(session, clsm->cursors);
	__wt_free(session, clsm->switch_txn);

	/* In case we were somehow left positioned, clear that. */
	__clsm_leave(clsm);

	/* The WT_LSM_TREE owns the URI. */
	cursor->uri = NULL;
	if (clsm->lsm_tree != NULL)
		__wt_lsm_tree_release(session, clsm->lsm_tree);
	WT_TRET(__wt_cursor_close(cursor));

err:	API_END_RET(session, ret);
}

/*
 * __wt_clsm_open --
 *	WT_SESSION->open_cursor method for LSM cursors.
 */
int
__wt_clsm_open(WT_SESSION_IMPL *session,
    const char *uri, WT_CURSOR *owner, const char *cfg[], WT_CURSOR **cursorp)
{
	WT_CONFIG_ITEM cval;
	WT_CURSOR_STATIC_INIT(iface,
	    __wt_cursor_get_key,	/* get-key */
	    __wt_cursor_get_value,	/* get-value */
	    __wt_cursor_set_key,	/* set-key */
	    __wt_cursor_set_value,	/* set-value */
	    __clsm_compare,		/* compare */
	    __wt_cursor_equals,		/* equals */
	    __clsm_next,		/* next */
	    __clsm_prev,		/* prev */
	    __clsm_reset,		/* reset */
	    __clsm_search,		/* search */
	    __clsm_search_near,		/* search-near */
	    __clsm_insert,		/* insert */
	    __clsm_update,		/* update */
	    __clsm_remove,		/* remove */
	    __wt_cursor_reconfigure,	/* reconfigure */
	    __wt_clsm_close);		/* close */
	WT_CURSOR *cursor;
	WT_CURSOR_LSM *clsm;
	WT_DECL_RET;
	WT_LSM_TREE *lsm_tree;
	bool bulk;

	clsm = NULL;
	cursor = NULL;
	lsm_tree = NULL;

	if (!WT_PREFIX_MATCH(uri, "lsm:"))
		return (EINVAL);

	if (F_ISSET(S2C(session), WT_CONN_IN_MEMORY))
		WT_RET_MSG(session, EINVAL,
		    "LSM trees not supported by in-memory configurations");

	WT_RET(__wt_config_gets_def(session, cfg, "checkpoint", 0, &cval));
	if (cval.len != 0)
		WT_RET_MSG(session, EINVAL,
		    "LSM does not support opening by checkpoint");

	WT_RET(__wt_config_gets_def(session, cfg, "bulk", 0, &cval));
	bulk = cval.val != 0;

	/* Get the LSM tree. */
	WT_WITH_HANDLE_LIST_LOCK(session,
	    ret = __wt_lsm_tree_get(session, uri, bulk, &lsm_tree));
	/*
	 * Check whether the exclusive open for a bulk load succeeded, and
	 * if it did ensure that it's safe to bulk load into the tree.
	 */
	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");
	/* Flag any errors from the tree get. */
	WT_ERR(ret);

	/* Make sure we have exclusive access if and only if we want it */
	WT_ASSERT(session, !bulk || lsm_tree->exclusive);

	WT_ERR(__wt_calloc_one(session, &clsm));

	cursor = &clsm->iface;
	*cursor = iface;
	cursor->session = &session->iface;
	cursor->uri = lsm_tree->name;
	cursor->key_format = lsm_tree->key_format;
	cursor->value_format = lsm_tree->value_format;

	clsm->lsm_tree = lsm_tree;

	/*
	 * The tree's dsk_gen starts at one, so starting the cursor on zero
	 * will force a call into open_cursors on the first operation.
	 */
	clsm->dsk_gen = 0;

	WT_STATIC_ASSERT(offsetof(WT_CURSOR_LSM, iface) == 0);
	WT_ERR(__wt_cursor_init(cursor, cursor->uri, owner, cfg, cursorp));

	if (bulk)
		WT_ERR(__wt_clsm_open_bulk(clsm, cfg));

	if (0) {
err:		if (clsm != NULL)
			WT_TRET(__wt_clsm_close(cursor));
		else if (lsm_tree != NULL)
			__wt_lsm_tree_release(session, lsm_tree);

		/*
		 * We open bulk cursors after setting the returned cursor.
		 * Fix that here.
		 */
		*cursorp = NULL;
	}

	return (ret);
}