je/dbi/CursorImpl.java

/*
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 2002, 2014 Oracle and/or its affiliates.  All rights reserved.
 *
 */

package com.sleepycat.je.dbi;

import java.util.Arrays;
import java.util.ArrayList;
import java.util.List;
import java.util.Comparator;
import java.util.Set;
import java.util.logging.Level;

import com.sleepycat.je.CacheMode;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.DuplicateDataException;
import com.sleepycat.je.EnvironmentFailureException;
import com.sleepycat.je.LockConflictException;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.evictor.Evictor.EvictionSource;
import com.sleepycat.je.latch.LatchSupport;
import com.sleepycat.je.log.LogUtils;
import com.sleepycat.je.log.ReplicationContext;
import com.sleepycat.je.tree.BIN;
import com.sleepycat.je.tree.BINBoundary;
import com.sleepycat.je.tree.IN;
import com.sleepycat.je.tree.Key;
import com.sleepycat.je.tree.LN;
import com.sleepycat.je.tree.SearchResult;
import com.sleepycat.je.tree.TrackingInfo;
import com.sleepycat.je.tree.Tree;
import com.sleepycat.je.tree.TreeWalkerStatsAccumulator;
import com.sleepycat.je.txn.LockGrantType;
import com.sleepycat.je.txn.LockInfo;
import com.sleepycat.je.txn.LockManager;
import com.sleepycat.je.txn.LockResult;
import com.sleepycat.je.txn.LockType;
import com.sleepycat.je.txn.Locker;
import com.sleepycat.je.txn.LockerFactory;
import com.sleepycat.je.txn.WriteLockInfo;
import com.sleepycat.je.utilint.DbLsn;
import com.sleepycat.je.utilint.LoggerUtils;
import com.sleepycat.je.utilint.Pair;
import com.sleepycat.je.utilint.StatGroup;
import com.sleepycat.je.utilint.TestHook;
import com.sleepycat.je.utilint.TestHookExecute;
import com.sleepycat.je.utilint.ThroughputStatGroup;
import com.sleepycat.je.utilint.VLSN;

/**
 * A CursorImpl is the internal implementation of the cursor.
 */
public class CursorImpl implements Cloneable {

    private static final boolean DEBUG = false;

    private static final byte CURSOR_NOT_INITIALIZED = 1;
    private static final byte CURSOR_INITIALIZED = 2;
    private static final byte CURSOR_CLOSED = 3;
    private static final String TRACE_DELETE = "Delete";
    private static final String TRACE_MOD = "Mod:";
    private static final String TRACE_INSERT = "Ins:";

    public static final int FOUND = 0x1;
    /* Exact match on the key portion. */
    public static final int EXACT_KEY = 0x2;
    /* Record found is the last one in the databaseImpl. */
    public static final int FOUND_LAST = 0x4;

    /*
     * Allocate hashCode ids from this. [#13896]
     */
    private static long lastAllocatedId = 0;

    /*
     * Unique id that we can return as a hashCode to prevent calls to
     * Object.hashCode(). [#13896]
     */
    private final int thisId;

    /* The databaseImpl behind the handle. */
    private final DatabaseImpl databaseImpl;

    /* Owning transaction. */
    private Locker locker;

    private final boolean retainNonTxnLocks;

    private final boolean isSecondaryCursor;

    /*
     * Cursor location in the databaseImpl, represented by a BIN and an index
     * in the BIN.  The bin is null if not established, and the index is
     * negative if not established.
     */
    private volatile BIN bin;
    private volatile int index;

    /* State of the cursor. See CURSOR_XXX above. */
    private byte status;

    private CacheMode cacheMode;
    private boolean allowEviction;

    /*
     * A cache of the record version for the operation at the current position.
     * Is null if the cursor is uninitialized.  For a secondary cursor, is the
     * version of the primary record.
     */
    private RecordVersion currentRecordVersion;

    private ThreadLocal<TreeWalkerStatsAccumulator> treeStatsAccumulatorTL;

    private TestHook testHook;

    public enum SearchMode {
        SET(true, false, "SET"),
        BOTH(true, true, "BOTH"),
        SET_RANGE(false, false, "SET_RANGE"),
        BOTH_RANGE(false, true, "BOTH_RANGE");

        private final boolean exactSearch;
        private final boolean dataSearch;
        private final String name;

        private SearchMode(boolean exactSearch,
                           boolean dataSearch,
                           String name) {
            this.exactSearch = exactSearch;
            this.dataSearch = dataSearch;
            this.name = "SearchMode." + name;
        }

        /**
         * Returns true when the key or key/data search is exact, i.e., for SET
         * and BOTH.
         */
        public final boolean isExactSearch() {
            return exactSearch;
        }

        /**
         * Returns true when the data value is included in the search, i.e.,
         * for BOTH and BOTH_RANGE.
         */
        public final boolean isDataSearch() {
            return dataSearch;
        }

        @Override
        public String toString() {
            return name;
        }
    }

    /**
     * Creates a cursor with retainNonTxnLocks=true, isSecondaryCursor=false.
     * These are the standard settings for an internal cursor.
     */
    public CursorImpl(DatabaseImpl database, Locker locker) {
        this(database, locker,
             true /*retainNonTxnLocks*/,
             false /*isSecondaryCursor*/);
    }

    /**
     * Creates a cursor.
     *
     * A cursor always retains transactional locks when it is reset or closed.
     * Non-transaction locks may be retained or not, depending on the
     * retainNonTxnLocks parameter value.
     *
     * Normally a user-created non-transactional Cursor releases locks on reset
     * and close, and a ThreadLocker is normally used.  However, by passing
     * true for retainNonTxnLocks a ThreadLocker can be made to retain locks;
     * this capability is used by SecondaryCursor.readPrimaryAfterGet.
     *
     * For internal (non-user) cursors, a BasicLocker is often used and locks
     * are retained. In these internal use cases the caller explicitly calls
     * BasicLocker.operationEnd() after the cursor is closed, and
     * retainNonTxnLocks is set to true to prevent the locks acquired by the
     * BasicLocker from being released when the cursor is closed.
     *
     * BasicLocker is also used for NameLN operations while opening a Database
     * handle.  Database handle locks must be retained, even if the Database is
     * opened non-transactionally.
     *
     * @param retainNonTxnLocks is true if non-transactional locks should be
     * retained (not released automatically) when the cursor is reset or
     * closed.
     *
     * @param isSecondaryCursor whether to treat this cursor as a secondary
     * cursor, e.g., secondary records don't have record versions.
     */
    public CursorImpl(
        DatabaseImpl databaseImpl,
        Locker locker,
        boolean retainNonTxnLocks,
        boolean isSecondaryCursor) {

        thisId = (int) getNextCursorId();
        bin = null;
        index = -1;

        this.retainNonTxnLocks = retainNonTxnLocks;
        this.isSecondaryCursor = isSecondaryCursor;

        /* Associate this cursor with the databaseImpl. */
        this.databaseImpl = databaseImpl;
        this.locker = locker;
        this.locker.registerCursor(this);

        /*
         * This default value is used only when the CursorImpl is used directly
         * (mainly for internal databases).  When the CursorImpl is created by
         * a Cursor, CursorImpl.setCacheMode will be called.
         */
        this.cacheMode = CacheMode.DEFAULT;

        status = CURSOR_NOT_INITIALIZED;

        /*
         * Do not perform eviction here because we may be synchronized on the
         * Database instance. For example, this happens when we call
         * Database.openCursor().  Also eviction may be disabled after the
         * cursor is constructed.
         */
    }

    /**
     * Performs a shallow copy and returns the new cursor.
     *
     * @param samePosition If true, this cursor's position is used for the new
     * cursor, and addCursor is called on the new cursor to register it with
     * the current BIN.  If false, the new cursor will be uninitialized.
     */
    public CursorImpl cloneCursor(final boolean samePosition) {

        assert assertCursorState(
            false /*mustBeInitialized*/, false /*mustNotBeInitialized*/);

        CursorImpl ret = null;
        try {
            latchBIN();

            ret = (CursorImpl) super.clone();

            if (!retainNonTxnLocks) {
                ret.locker = locker.newNonTxnLocker();
            }

            ret.locker.registerCursor(ret);

            if (samePosition) {
                ret.addCursor();
            } else {
                ret.clear();
            }
        } catch (CloneNotSupportedException cannotOccur) {
            return null;
        } finally {
            releaseBIN();
        }

        /* Perform eviction before and after each cursor operation. */
        criticalEviction();

        return ret;
    }

    /*
     * Allocate a new hashCode id.  Doesn't need to be synchronized since it's
     * ok for two objects to have the same hashcode.
     */
    private static long getNextCursorId() {
        return ++lastAllocatedId;
    }

    @Override
    public int hashCode() {
        return thisId;
    }

    public Locker getLocker() {
        return locker;
    }

    /**
     * Called when a cursor has been duplicated prior to being moved.  The new
     * locker is informed of the old locker, so that a preempted lock taken by
     * the old locker can be ignored. [#16513]
     *
     * @param closingCursor the old cursor that will be closed if the new
     * cursor is moved successfully.
     */
    public void setClosingLocker(CursorImpl closingCursor) {

        /*
         * If the two lockers are different, then the old locker will be closed
         * when the operation is complete.  This is currently the case only for
         * ReadCommitted cursors and non-transactional cursors that do not
         * retain locks.
         */
        if (!retainNonTxnLocks && locker != closingCursor.locker) {
            locker.setClosingLocker(closingCursor.locker);
        }
    }

    /**
     * Called when a cursor move operation is complete.  Clears the
     * closingLocker so that a reference to the old closed locker is not held.
     */
    public void clearClosingLocker() {
        locker.setClosingLocker(null);
    }

    public CacheMode getCacheMode() {
        return cacheMode;
    }

    /**
     * Sets the effective cache mode to use for the next operation.  The
     * cacheMode field will never be set to null or DYNAMIC, and can be passed
     * directly to latching methods.
     *
     * @see #performCacheEviction
     */
    public void setCacheMode(final CacheMode mode) {
        cacheMode = databaseImpl.getEffectiveCacheMode(mode);
    }

    public void setTreeStatsAccumulator(TreeWalkerStatsAccumulator tSA) {
        maybeInitTreeStatsAccumulator();
        treeStatsAccumulatorTL.set(tSA);
    }

    private void maybeInitTreeStatsAccumulator() {
        if (treeStatsAccumulatorTL == null) {
            treeStatsAccumulatorTL =
                new ThreadLocal<TreeWalkerStatsAccumulator>();
        }
    }

    private TreeWalkerStatsAccumulator getTreeStatsAccumulator() {
        if (EnvironmentImpl.getThreadLocalReferenceCount() > 0) {
            maybeInitTreeStatsAccumulator();
            return treeStatsAccumulatorTL.get();
        } else {
            return null;
        }
    }

    public void incrementLNCount() {
        TreeWalkerStatsAccumulator treeStatsAccumulator =
            getTreeStatsAccumulator();
        if (treeStatsAccumulator != null) {
            treeStatsAccumulator.incrementLNCount();
        }
    }

    public int getIndex() {
        return index;
    }

    public BIN getBIN() {
        return bin;
    }

    public void setIndex(int idx) {
        index = idx;
    }

    public void setOnFirstSlot() {
        assert(bin.isLatchOwner());
        index = 0;
    }

    public void setOnLastSlot() {
        assert(bin.isLatchOwner());
        index = bin.getNEntries() - 1;
    }

    public boolean isOnBIN(BIN bin) {
        return this.bin == bin;
    }

    public void assertBIN(BIN bin) {
        assert this.bin == bin :
            "nodeId=" + bin.getNodeId() +
            " cursor=" + dumpToString(true);
    }

    public boolean isOnSamePosition(CursorImpl other) {
        return bin == other.bin && index == other.index;
    }

    public void setBIN(BIN newBin) {

        /*
         * Historical note. In the past we checked here that the cursor was
         * removed for the prior BIN by calling BIN.containsCursor [#16280].
         * Because the containsCursor method takes a latch on the prior BIN,
         * this causes a rare latch deadlock when newBin is latched (during an
         * insert, for example), since this thread will latch two BINs in
         * arbitrary order; so the assertion was removed [#21395].
         */
        bin = newBin;
    }

    public void latchBIN()
        throws DatabaseException {

        while (bin != null) {
            BIN waitingOn = bin;
            waitingOn.latch(cacheMode);
            if (bin == waitingOn) {
                return;
            }
            waitingOn.releaseLatch();
        }
    }

    public void releaseBIN() {
        if (bin != null) {
            bin.releaseLatchIfOwner();
        }
    }

    void addCursor(BIN bin) {
        if (bin != null) {
            assert bin.isLatchExclusiveOwner();
            bin.addCursor(this);
        }
    }

    /**
     * Add to the current cursor.
     */
    void addCursor() {
        if (bin != null) {
            addCursor(bin);
        }
    }

    /**
     * Change cursor to point to the given BIN/index.  If the new BIN is
     * different, then old BIN must be unlatched and the new BIN must be
     * latched.
     */
    private void setPosition(BIN newBin, int newIndex) {
        if (bin != newBin) {
            if (bin != null) {
                latchBIN();
                bin.removeCursor(this);
                bin.releaseLatch();
            }
            setBIN(newBin);
            addCursor();
        }
        setIndex(newIndex);
    }

    /**
     * Called for creating trace messages without any latching.
     */
    public long getCurrentNodeId() {
        final BIN b = bin;
        return (b == null ? -1 : b.getNodeId());
    }

    public long getCurrentLsn() {

        assert(bin != null && bin.isLatchOwner());
        assert(index >= 0 && index < bin.getNEntries());

        return bin.getLsn(index);
    }

    public byte[] getCurrentKey() {
        return getCurrentKey(false);
    }

    /**
     * Returns the key at the current position, regardless of whether the
     * record is deleted.  Does not lock. The key returned is not a copy and
     * may not be returned directly to the user without copying it first.
     *
     * The cursor must be initialized.
     */
    public byte[] getCurrentKey(boolean isLatched) {

        if (!isLatched) {
            latchBIN();
        }

        try {
            assert(bin != null);
            assert(index >= 0 && index < bin.getNEntries());

            return bin.getKey(index);
        } finally {
            if (!isLatched) {
                releaseBIN();
            }
        }
    }

    private void setInitialized() {
        status = CURSOR_INITIALIZED;
    }

    /**
     * @return true if this cursor is closed
     */
    public boolean isClosed() {
        return (status == CURSOR_CLOSED);
    }

    /**
     * @return true if this cursor is not initialized
     */
    public boolean isNotInitialized() {
        return (status == CURSOR_NOT_INITIALIZED);
    }

    public boolean isInternalDbCursor() {
        return databaseImpl.isInternalDb();
    }

    public boolean hasDuplicates() {
        return databaseImpl.getSortedDuplicates();
    }

    public void dumpTree() {
        databaseImpl.getTree().dump();
    }

    /**
     * For a non-sticky cursor, this is an alternative to closing or resetting
     * the cursor when it is initialized and an advancing operation
     * (next/prev/skip) is about to be performed. The cursor position cannot be
     * reset as it would be if the operation were a search, for example.
     */
    public void beforeAdvance() {

        /*
         * When EVICT_LN is configured we should evict the LN at the current
         * position before changing the position. [#21973]
         *
         * Note that EVICT_BIN cannot be implemented for a non-sticky cursor.
         * We would have to call removeCursor before the BIN eviction, and then
         * our position would be unstable. Even if this were possible, since
         * we can't predict the new position we would evict the BIN at every
         * operation, even when the BIN remained the same.
         *
         * It might be possible to support MAKE_COLD here, but more thought and
         * testing would be required.
         *
         * Therefore, when EVICT_BIN or MAKE_COLD are configured, cursor
         * cloning is always used.
         */
        if (cacheMode == CacheMode.EVICT_LN && bin != null) {
            evict();
        }

        releaseNonTxnLocks();

        criticalEviction();
    }

    /**
     * Reset a cursor to an uninitialized state, but unlike close(), allow it
     * to be used further.
     */
    public void reset()
        throws DatabaseException {

        /* Must remove cursor before evicting BIN and releasing locks. */
        removeCursorAndPerformCacheEviction(null /*newCursor*/);

        releaseNonTxnLocks();

        /* Perform eviction before and after each cursor operation. */
        criticalEviction();
    }

    private void clear() {
        bin = null;
        index = -1;
        status = CURSOR_NOT_INITIALIZED;
        currentRecordVersion = null;
    }

    private void releaseNonTxnLocks() {
        if (!retainNonTxnLocks) {
            locker.releaseNonTxnLocks();
        }
    }

    public void close()
        throws DatabaseException {

        close(null);
    }

    /**
     * Close a cursor.
     *
     * @param newCursor is another cursor that is kept open by the parent
     * Cursor object, or null if no other cursor is kept open.
     *
     * @throws DatabaseException if the cursor was previously closed.
     */
    public void close(final CursorImpl newCursor)
        throws DatabaseException {

        assert assertCursorState(
            false /*mustBeInitialized*/, false /*mustNotBeInitialized*/);

        /* Must remove cursor before evicting BIN and releasing locks. */
        removeCursorAndPerformCacheEviction(newCursor);

        locker.unRegisterCursor(this);

        if (!retainNonTxnLocks) {
            locker.nonTxnOperationEnd();
        }

        status = CURSOR_CLOSED;

        /* Perform eviction before and after each cursor operation. */
        criticalEviction();
    }

    private void removeCursorAndPerformCacheEviction(CursorImpl newCursor) {

        latchBIN();

        if (bin == null) {
            clear(); // ensure that state is uninitialized
            return;
        }

        try {
            /* Must remove cursor before evicting BIN. */
            bin.removeCursor(this);
            performCacheEviction(newCursor); // may release latch
        } finally {
            releaseBIN();
            clear();
        }
    }

    /**
     * Disables or enables eviction during cursor operations.  For example, a
     * cursor used to implement eviction (e.g., in some UtilizationProfile and
     * most DbTree and VLSNIndex methods) should not itself perform eviction,
     * but eviction should be enabled for user cursors.  Eviction is disabled
     * by default.
     */
    public void setAllowEviction(boolean allowed) {
        allowEviction = allowed;
    }

    public void criticalEviction() {

        /*
         * In addition to disabling critical eviction for internal cursors (see
         * setAllowEviction above), we do not perform critical eviction when
         * EVICT_BIN or MAKE_COLD is used.  Operations using MAKE_COLD and
         * EVICT_BIN generally do not add any net memory to the cache, so they
         * shouldn't have to perform critical eviction or block while another
         * thread performs eviction.
         */
        if (allowEviction &&
            cacheMode != CacheMode.MAKE_COLD &&
            cacheMode != CacheMode.EVICT_BIN) {
            databaseImpl.getEnv().criticalEviction(false /*backgroundIO*/);
        }
    }

    /**
     * When multiple operations are performed, CacheMode-based eviction is
     * performed for a given operation at the end of the next operation, which
     * calls close() or reset() on the CursorImpl of the previous operation.
     * Eviction for the last operation (including when only one operation is
     * performed) also occurs during Cursor.close(), which calls
     * CursorImpl.close().
     *
     * By default, the CacheMode returned by DatabaseImpl.getCacheMode is used,
     * and the defaults specified by the user for the Database or Environment
     * are applied.  However, the default mode can be overridden by the user by
     * calling Cursor.setCacheMode, and the mode may be changed prior to each
     * operation, if desired.
     *
     * To implement a per-operation CacheMode, two CacheMode fields are
     * maintained.  Cursor.cacheMode is the mode to use for the next operation.
     * CursorImpl.cacheMode is the mode that was used for the previous
     * operation, and that is used for eviction when that CursorImpl is closed
     * or reset.
     *
     * This method must be called with the BIN latched but may release it,
     * namely when the BIN is evicted.
     */
    private void performCacheEviction(final CursorImpl newCursor) {

        EnvironmentImpl envImpl = databaseImpl.getEnv();

        if (cacheMode != CacheMode.EVICT_LN &&
            cacheMode != CacheMode.EVICT_BIN &&
            (cacheMode != CacheMode.MAKE_COLD ||
             !(envImpl.isCacheFull() || envImpl.wasCacheEverFull()))) {
            /* Eviction not configured, or for MAKE_COLD, cache never full. */
            return;
        }

        final BIN nextBin;
        final int nextIndex;

        if (newCursor != null) {
            nextBin = newCursor.bin;
            nextIndex = newCursor.index;
        } else {
            nextBin = null;
            nextIndex = -1;
        }

        switch (cacheMode) {
        case EVICT_LN:
            /* Evict the LN if we've moved to a new record. */
            if (bin != nextBin || index != nextIndex) {
                evict(true);
            }
            break;
        case EVICT_BIN:
            if (bin == nextBin) {
                /* BIN has not changed, do not evict it.  May evict the LN. */
                if (index != nextIndex) {
                    evict(true);
                }
            } else {
                /* BIN has changed, evict it. */
                evictBIN();
            }
            break;
        case MAKE_COLD:
            if (bin == nextBin) {
                /* BIN has not changed, do not evict it.  May evict the LN. */
                if (index != nextIndex) {
                    evict(true);
                }
            } else if (!envImpl.isCacheFull()) {
                /*
                 * The BIN has changed, but the cache is not full. Just evict
                 * the LN.
                 */
                evict(true);
            } else {
                /* BIN has changed, evict it. */
                evictBIN();
            }
            break;
        default:
            assert false;
        }
    }

    /**
     * Evict current BIN.  Must already be latched. The latch will be released
     * inside the doEvictOneIN() call.
     */
    private void evictBIN() {
        databaseImpl.getEnv().getEvictor().doEvictOneIN(
            bin, EvictionSource.CACHEMODE);
    }

    /**
     * Evict the LN node at the cursor position.
     */
    public void evict()
        throws DatabaseException {

        evict(false); // isLatched
    }

    /**
     * Evict the LN node at the cursor position.
     */
    public void evict(boolean isLatched)
        throws DatabaseException {

        try {
            if (!isLatched) {
                latchBIN();
            }
            if (index >= 0) {
                bin.evictLN(index);
            }
        } finally {
            if (!isLatched) {
                releaseBIN();
            }
        }
    }

    public OperationStatus deleteCurrentRecord(
        ReplicationContext repContext)
        throws DatabaseException {
        return deleteCurrentRecord(repContext, null);
    }

    /**
     * Delete the item pointed to by the cursor. If the item is already
     * deleted, return KEYEMPTY. Returns with nothing latched.
     *
     * @param opStats Used to collect stats about user ops performed on BIN
     * deltas. Internal ops should pass null.
     *
     * @return 0 on success, appropriate error code otherwise.
     */
    public OperationStatus deleteCurrentRecord(
        ReplicationContext repContext,
        ThroughputStatGroup opStats)
        throws DatabaseException {

        assert assertCursorState(
            true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);
        final EnvironmentImpl envImpl = databaseImpl.getEnv();
        final DbType dbType = databaseImpl.getDbType();

        boolean success = false;
        final long oldLsn;
        final LN.LogResult logResult;

        latchBIN();

        try {
            /*
             * Get a write lock.  An uncontended lock is permitted because we
             * will log a new LN before releasing the BIN latch.
             */
            final LockStanding lockStanding = lockLN(
                LockType.WRITE, true /*allowUncontended*/, false /*noWait*/);

            if (!lockStanding.recordExists()) {
                /* LN was deleted or cleaned. */
                revertLock(lockStanding);
                success = true;
                return OperationStatus.KEYEMPTY;
            }

            oldLsn = lockStanding.lsn;
            assert oldLsn != DbLsn.NULL_LSN;

            /*
             * Must fetch LN if any of the following are true:
             *  - CLEANER_FETCH_OBSOLETE_SIZE is configured and lastLoggedSize
             *    is unknown
             *  - this database does not use the standard LN class and we
             *    cannot call DbType.createdDeletedLN further below
             * For other cases, we are careful not to fetch, in order to avoid
             * a random read during a delete operation.
             *
             * Note that we checked for a deleted/cleaned LN above, so we know
             * that fetchLN will not return null.
             */
            final int lastLoggedSize = bin.getLastLoggedSize(index);
            LN ln;
            if ((lastLoggedSize == 0 &&
                 envImpl.getCleaner().getFetchObsoleteSize(databaseImpl)) ||
                !dbType.mayCreateDeletedLN()) {
                /* Must fetch. */
                ln = bin.fetchLN(index, cacheMode);
            } else {
                /* If resident, use LN for obsolete size tracking. */
                ln = (LN) bin.getTarget(index);
            }

            /*
             * Make the existing LN deleted, if cached, else create a new
             * deleted LN.
             */
            final long oldLNMemSize;
            if (ln != null) {
                /* Leave LN cached. Set ln.data to null and mark it dirty. */
                oldLNMemSize = ln.getMemorySizeIncludedByParent();
                ln.delete();
            } else {
                /*
                 * LN is not cached. Create an LN with ln.data == null, and
                 * mark it dirty. Then attach it to the tree.
                 */
                ln = dbType.createDeletedLN(envImpl);
                oldLNMemSize = ln.getMemorySizeIncludedByParent();
                bin.attachNode(index, ln, null /*lnSlotKey*/);
            }

            /* Get a wli to log. */
            WriteLockInfo wli = lockStanding.prepareForUpdate();

            /* Log the deleted record version and lock its new LSN. */
            logResult = ln.optionalLog(
                envImpl, databaseImpl, bin.getKey(index), oldLsn,
                lastLoggedSize, locker, wli, repContext);

            /*
             * Now update the parent BIN of the LN to correctly reference the
             * LN and adjust the memory sizing.
             */
            bin.updateEntry(
                index, oldLNMemSize, logResult.newLsn, logResult.newSize,
                null /*lnSlotKey*/);
            bin.setPendingDeleted(index);

            /* Cache record version for delete operation. */
            setCurrentVersion(ln.getVLSNSequence(), logResult.newLsn);

            /*
             * It is desirable to evict the LN after deletion because it will
             * never be fetched again.  This is especially true because slot
             * compression is deferred until a full BIN is logged.  But for
             * deferred-write DBs we should not evict a dirty LN since it may
             * be logged unnecessarily.
             */
            if (!databaseImpl.isDeferredWriteMode() &&
                bin.getTarget(index) != null) {
                bin.evictLN(index);
            }

            locker.addDeleteInfo(bin);
            success = true;

        } finally {

            if (success &&
                opStats != null &&
                bin != null &&
                bin.isBINDelta()) {
                opStats.increment(
                    ThroughputStatGroup.BIN_DELTA_DELETES_OFFSET);
            }

            releaseBIN();
        }

        trace(Level.FINER, TRACE_DELETE, bin, index, oldLsn, logResult.newLsn);

        return OperationStatus.SUCCESS;
    }

    /**
     * Modify the current record with the given data, and optionally replace
     * the key.
     *
     * @param key The new key value for the BIN slot S to be updated. Cannot
     * be partial. For a no-dups DB, it is null. For dups DBs it is a 2-part
     * key combining the current primary key of slot S with the original,
     * user-provided data. "key" (if not null) must compare equal to S.key
     * (otherwise DuplicateDataException is thrown), but the 2 keys may not
     * be identical if custom comparators are used. So, S.key will actually
     * be replaced by "key".
     *
     * @param data The new data to (perhaps partially) replace the data of the
     * LN associated with the BIN slot. For dups DBs it is EMPTY_DUPS_DATA.
     * Note: for dups DBs the original, user-provided "data" must not be
     * partial.
     *
     * @param returnOldData To receive the old LN data (before the update).
     * It is needed only by DBs with indexes/triggers; will be null otherwise.
     *
     * @param returnNewData To receive the full data of the updated LN.
     * It is needed only by DBs with indexes/triggers and only if "data" is
     * partial; will be null otherwise. Note: "returnNewData" may be different
     * than "data" only if "data" is partial.
     *
     * @param opStats Used to collect stats about user ops performed on BIN
     * deltas. Internal ops should pass null.
     */
    public OperationStatus updateCurrentRecord(
        DatabaseEntry key,
        DatabaseEntry data,
        DatabaseEntry returnOldData,
        DatabaseEntry returnNewData,
        ReplicationContext repContext,
        ThroughputStatGroup opStats) {

        assert assertCursorState(
            true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);

        if (returnOldData != null) {
            returnOldData.setData(null);
        }
        if (returnNewData != null) {
            returnNewData.setData(null);
        }

        final LockStanding lockStanding;
        OperationStatus status = OperationStatus.NOTFOUND;
        boolean success = false;

        latchBIN();

        try {
            /* Get a write lock. */
            lockStanding = lockLN(
                LockType.WRITE, true /*allowUncontended*/, false /*noWait*/);

            if (!lockStanding.recordExists()) {
                /* LN was deleted or cleaned. */
                revertLock(lockStanding);
            } else {
                status = updateRecordInternal(
                    (key != null ? Key.makeKey(key) : null), data,
                    returnOldData, returnNewData, lockStanding, repContext);
            }

            success = true;
            return status;

        } finally {

            if (success &&
                opStats != null &&
                bin != null &&
                bin.isBINDelta()) {
                opStats.increment(
                    ThroughputStatGroup.BIN_DELTA_UPDATES_OFFSET);
            }

            releaseBIN();
        }
    }

    /**
     * Insert the given record (key + LN) in the tree or return KEYEXIST if
     * the key is already present.
     *
     * The cursor must initially be uninitialized.
     *
     * This method is called directly internally for putting tree map LNs
     * and file summary LNs.
     */
    public OperationStatus insertRecord(
        byte[] key,
        LN ln,
        boolean blindInsertion,
        ReplicationContext repContext)
        throws DatabaseException {

        assert assertCursorState(
            false /*mustBeInitialized*/, true /*mustNotBeInitialized*/);
        if (LatchSupport.TRACK_LATCHES) {
            LatchSupport.expectBtreeLatchesHeld(0);
        }

        try {
            final Pair<LockStanding, Boolean> result = insertRecordInternal(
                key, ln, blindInsertion, null, repContext);

            if (!result.second()) {
                return OperationStatus.KEYEXIST;
            }
            return OperationStatus.SUCCESS;
        } finally {
            releaseBIN();
        }
    }

    /**
     * Insert or update a given record. The method searches for the record
     * using its key. It will perform an update if the record is found,
     * otherwise an insertion.
     *
     * The cursor must initially be uninitialized.
     *
     * Called by all the Cursor.putXXX() ops, except putCurrent().
     *
     * @param key The new key value for the BIN slot S to be inserted/updated.
     * Cannot be partial. For dups DBs it is a 2-part key combining the
     * original, user-provided key and data. In case of update, "key" must
     * compare equal to S.key (otherwise DuplicateDataException is thrown),
     * but the 2 keys may not be identical if custom comparators are used.
     * So, S.key will actually be replaced by "key".
     *
     * @param data In case of update, the new data to (perhaps partially)
     * replace the data of the LN associated with the BIN slot. For dups DBs
     * it is EMPTY_DUPS_DATA. Note: for dups DBs the original, user-provided
     * "data" must not be partial.
     *
     * @param ln is normally a new LN node that is created for insertion, and
     * will be discarded if an update occurs.  However, HA will pass an
     * existing node.
     *
     * @param putMode OVERWRITE or NO_OVERWRITE
     *
     * @param returnOldData To receive, in case of update, the old LN data
     * (before the update). It is needed only by DBs with indexes/triggers;
     * will be null otherwise.
     *
     * @param returnNewData To receive the full data of the new or updated LN.
     * It is needed only by DBs with indexes/triggers and only if "data" is
     * partial; will be null otherwise. Note: "returnNewData" may be different
     * than "data" only if "data" is partial.
     *
     * @param opStats Used to collect stats about user ops performed on BIN
     * deltas. Internal ops should pass null.
     *
     * @return pair of status and 'inserted' boolean.
     */
    public Pair<OperationStatus, Boolean> insertOrUpdateRecord(
        final DatabaseEntry key,
        final DatabaseEntry data,
        final LN ln,
        final PutMode putMode,
        final DatabaseEntry returnOldData,
        final DatabaseEntry returnNewData,
        final ReplicationContext repContext,
        ThroughputStatGroup opStats) {

        assert key != null;
        assert data != null;
        assert ln != null;
        assert putMode != null;
        assert assertCursorState(
            false /*mustBeInitialized*/, true /*mustNotBeInitialized*/);
        if (LatchSupport.TRACK_LATCHES) {
            LatchSupport.expectBtreeLatchesHeld(0);
        }

        if (putMode != PutMode.OVERWRITE &&
            putMode != PutMode.NO_OVERWRITE &&
            putMode != PutMode.BLIND_INSERTION) {
            throw EnvironmentFailureException.unexpectedState(
                putMode.toString());
        }

        boolean success = false;
        boolean inserted = false;

        byte[] keyCopy = Key.makeKey(key);

        try {

            /*
             * Try to insert the key/data pair as a new record. Will succeed if
             * the record does not exist in the DB already. Otherwise, the
             * insertRecord() returns with the cursor registered on the slot
             * whose key is equal to "key", with the LSN of that slot  ocked
             * in WRITE mode, and with the containing BIN latched.
             */
            Pair<LockStanding, Boolean> insertResult = insertRecordInternal(
                keyCopy, ln,
                (putMode == PutMode.BLIND_INSERTION) /*blindInsertion*/,
                returnNewData, repContext);

            if (insertResult.second()) {
                inserted = true;
                success = true;
                return new Pair<>(OperationStatus.SUCCESS, true);
            }

            /*
             * There is a non-deleted slot whose key is == "key". So, this is
             * going to be an update. Note: Cursor has been registered on the
             * existing slot by insertRecord()
             */
            if (putMode == PutMode.NO_OVERWRITE) {
                success = true;
                return new Pair<>(OperationStatus.KEYEXIST, false);
            }

            /*
             * Update the non-deleted record at the cursor position. We have
             * optimized by preferring to take an uncontended lock. The
             * lockStanding var is guaranteed to be non-null in this case.
             * The BIN must remain latched when calling this method.
             */
            final OperationStatus status = updateRecordInternal(
                keyCopy, data, returnOldData, returnNewData,
                insertResult.first(), repContext);

            success = true;
            return new Pair<>(status, false);

        } finally {

            if (success &&
                opStats != null &&
                bin != null &&
                bin.isBINDelta()) {
                if (inserted) {
                    opStats.increment(
                        ThroughputStatGroup.BIN_DELTA_INSERTS_OFFSET);
                } else {
                    opStats.increment(
                        ThroughputStatGroup.BIN_DELTA_UPDATES_OFFSET);
                }
            }

            releaseBIN();
        }
    }

    /*
     * Try to insert the key/data pair as a new record. Will succeed if the
     * record does not exist already.
     *
     * The cursor must initially be uninitialized.
     *
     * On return, this.bin is latched.
     */
    private Pair<LockStanding, Boolean> insertRecordInternal(
        final byte[] key,
        final LN ln,
        boolean blindInsertion,
        DatabaseEntry returnNewData,
        final ReplicationContext repContext) {

        final EnvironmentImpl envImpl = databaseImpl.getEnv();
        WriteLockInfo wli;
        LockStanding lockStanding = null;
        final boolean isSlotReuse;
        final Tree tree = databaseImpl.getTree();

        /*
         * At this point, this cursor does not have a position so it cannot be
         * registered with the BIN that will be used. This is good because it
         * allows slot compression to occur before BIN splits (thus avoiding
         * splits if compression finds and removes any deleted slots). However,
         * if another cursor, including the one from which this was cloned, is
         * registered with the BIN, then splits won't be allowed. This is a
         * good reason to use non-sticky cursors for insertions, especially
         * sequential insertions since they will often end up in the same BIN.
         *
         * Find and latch the BIN that should contain the "key". On return from
         * the tree search, this.bin is latched, but "this" is still not
         * registered.
         */
        bin = tree.findBinForInsert(key, getCacheMode());

        /*
         * In the case where logging occurs before locking, allow lockers to
         * reject the operation (e.g., if writing on a replica) and also
         * prepare to undo in the (very unlikely) event that logging succeeds
         * but locking fails. Call this method BEFORE slot insertion, in case
         * it throws an exception which would leave the slot with a null LSN.
         *
         * For Txn, creates the writeInfo map (if not done already), and
         * inserts databaseImpl in the undoDatabases map. Noop for other
         * non-HA lockers.
         */
        locker.preLogWithoutLock(databaseImpl);

        /*
         * If the key exists already, insertEntry1() does not insert, but
         * returns the index of the existing key.
         *
         * If bin is a delta and it does not contain the key, then:
         * (a) if blindInsertion is false, insertEntry1() will mutate it to a
         * full BIN and check again if the key exists or not.
         * (b) if blindInsertion is true, insertEntry1() will not mutate the
         * delta; it will just insert the key into the delta. This is OK,
         * because blindInsertion will be true only if we know already that the
         * key does not exist in the tree.
         */
        int insertIndex = bin.insertEntry1(
            ln, key, DbLsn.NULL_LSN, blindInsertion);

        if ((insertIndex & IN.INSERT_SUCCESS) == 0) {
            /*
             * Key exists. Insertion was not successful. Register the cursor on
             * the existing slot. If the slot is marked deleted, the key does
             * not really exist and the slot can be reused to do an insertion.
             */
            isSlotReuse = true;
            setIndex(insertIndex);
            addCursor();
            setInitialized();

            /*
             * Lock the LSN for the existing LN slot, and check deleted-ness.
             * An uncontended lock request is permitted because we are holding
             * the bin latch. If no locker holds a lock on the slot, then no
             * lock is taken by this cursor either.
             */
            lockStanding = lockLN(
                LockType.WRITE, true /*allowUncontended*/, false /*noWait*/);

            boolean isDeleted = !lockStanding.recordExists();

            if (!isDeleted)
                return new Pair<>(lockStanding, false);

            /*
             * The current slot has its KD or PD flag set. Note: the record of
             * the slot may have been deleted by this.locker itself.
             */
            assert(lockStanding != null);

            /*
             * Create a new WriteLockInfo or get an existing one for the LSN
             * of the current slot, and set its abortLSN and abortKD fields,
             * if needed, i.e, if it is not the current txn the one who created
             * this LSN. The abortLSN and abortKD fields of the wli will be
             * included in the new logrec.
             */
            wli = lockStanding.prepareForUpdate();

        } else {
            /*
             * Register the cursor at the slot that has been successfully
             * inserted.
             */
            isSlotReuse = false;
            setIndex(insertIndex &= ~IN.INSERT_SUCCESS);
            addCursor();
            setInitialized();

            /* Create a new WriteLockInfo */
            wli = LockStanding.prepareForInsert();
        }

        /*
         * Log the new LN and lock the LSN of the new logrec in WRITE mode.
         * Note: in case of slot reuse, we pass NULL_LSN for the oldLsn param
         * because the old LN was counted obsolete when it was deleted.
         */
        LN.LogResult logResult = null;
        try {
            logResult = ln.optionalLog(
                envImpl, databaseImpl, key,
                DbLsn.NULL_LSN /*oldLSN*/, 0 /*oldSize*/,
                locker, wli, repContext);
        } finally {
            if (logResult == null && !isSlotReuse) {
                /*
                 * Possible buffer overflow, out-of-memory, or I/O exception
                 * during logging. The BIN entry will contain a NULL_LSN. To
                 * prevent an exception during a future fetchLN() call, we
                 * set the KD flag. We do not call BIN.deleteEntry because it
                 * does not adjust cursors. We do not add this entry to the
                 * compressor queue to avoid complexity (this situation is
                 * rare).
                 */
                bin.setKnownDeleted(index);
            }
        }

        if (lockStanding == null) {
            /*
             * No slot reuse; straight insertion. Update LSN in BIN slot.
             * The LN is already in the slot.
             */
            bin.updateEntry(index, logResult.newLsn, logResult.newSize);

            /*
             * The following call accounts for extra marshaled memory, i.e.,
             * memory that was added to the LN as a side-effect of logging it.
             * This can happen for FileSummaryLN's only (it is a noop for
             * other kinds of LNs).
             *
             * To avoid violating assertions (e.g., in IN.changeMemorySize), we
             * must must finish the memory adjustment while the BIN is still
             * latched. [#20069]
             *
             * This special handling does not apply to slot reuse, because the
             * updateEntry() version used in the slot reuse case will recalc
             * the BIN memory from scratch, and as a result, will take into
             * account the extra marshaled memory. [#20845]
             */
            ln.addExtraMarshaledMemorySize(bin);
        } else {

            /* For DW DBs, set the slot LSN to NULL. Otherwise, noop. */
            bin.prepareForSlotReuse(index);

            /*
             * Slot reuse. When reusing a slot, the key is replaced in the BIN
             * slot. This ensures that the correct key value is used when the
             * new key is non-identical to the key in the slot but is
             * considered equal by the btree comparator.
             */
            bin.updateEntry(index, ln, logResult.newLsn, logResult.newSize, key);
            bin.clearKnownDeleted(index);
            bin.clearPendingDeleted(index);
        }

        if (returnNewData != null) {
            returnNewData.setData(null);
            ln.setEntry(returnNewData);
        }

        /* Cursor is positioned on new record. */
        setInitialized();

        /* Cache record version for insertion operation. */
        setCurrentVersion(ln.getVLSNSequence(), bin.getLsn(index));

        /*
         * It is desirable to evict the LN in a duplicates DB because it will
         * never be fetched again. But for deferred-write DBs we should not
         * evict a dirty LN since it may be logged unnecessarily.
         */
        if (databaseImpl.getSortedDuplicates() &&
            !databaseImpl.isDeferredWriteMode() &&
            bin.getTarget(index) != null) {
            bin.evictLN(index);
        }

        traceInsert(Level.FINER, bin, logResult.newLsn, index);

        return new Pair<>(lockStanding, true);
    }

    /**
     * Update the record where the cursor is currently positioned at. The
     * cursor is registered with this position, the associated bin is latched,
     * the BIN slot is not marked deleted, and it has been locked in WRITE
     * mode.
     */
    private OperationStatus updateRecordInternal(
        byte[] key,
        DatabaseEntry data,
        DatabaseEntry returnOldData,
        DatabaseEntry returnNewData,
        LockStanding lockStanding,
        ReplicationContext repContext) {

        final EnvironmentImpl envImpl = databaseImpl.getEnv();
        final DbType dbType = databaseImpl.getDbType();

        assert lockStanding.recordExists();
        final long oldLsn = lockStanding.lsn;
        assert oldLsn != DbLsn.NULL_LSN;

        final LN.LogResult logResult;

        /*
         * Must fetch LN if any of the following are true:
         *  - returnOldData is non-null: data needs to be returned
         *  - data is a partial entry: needs to be resolved
         *  - CLEANER_FETCH_OBSOLETE_SIZE is configured and lastLoggedSize
         *    is unknown
         *  - this database does not use the standard LN class and we
         *    cannot call DbType.createdUpdatedLN further below (this is
         *    the case for NameLNs, MapLNs, and FileSummaryLNs).
         * For other cases, we are careful not to fetch, in order to avoid
         * a random read during an update operation.
         *
         * Note that we checked for a deleted/cleaned LN above, so we know
         * that fetchLN will not return null.
         */
        final int lastLoggedSize = bin.getLastLoggedSize(index);
        LN ln;

        if (returnOldData != null ||
            data.getPartial() ||
            (lastLoggedSize == 0 &&
             envImpl.getCleaner().getFetchObsoleteSize(databaseImpl)) ||
            !dbType.mayCreateUpdatedLN()) {
            /* Must fetch. */
            ln = bin.fetchLN(index, cacheMode);
        } else {
            /* If resident, use LN for obsolete size tracking. */
            ln = (LN) bin.getTarget(index);
        }

        final byte[] oldKey = bin.getKey(index);
        final byte[] foundDataBytes = (ln != null) ? ln.getData() : null;
        final byte[] newData = (data.getPartial() ?
                                LN.resolvePartialEntry(data, foundDataBytes) :
                                LN.copyEntryData(data));

        /*
         * If the key is changed (according to the comparator), we assume
         * it is actually the data that has changed for a duplicate's DB.
         */
        if (key != null &&
            Key.compareKeys(
                oldKey, key, databaseImpl.getKeyComparator()) != 0) {
            throw new DuplicateDataException(
                "Can't replace a duplicate with new data that is not " +
                "equal to the existing data according to the duplicate " +
                " comparator.");
        }

        if (returnOldData != null) {
            assert foundDataBytes != null;
            returnOldData.setData(null);
            LN.setEntry(returnOldData, foundDataBytes);
        }

        /* Update the existing LN, if cached, else create new LN. */
        final long oldLNMemSize;
        if (ln != null) {
            /* LN is cached: set ln.data to newData and mark ln dirty. */
            oldLNMemSize = ln.getMemorySizeIncludedByParent();
            ln.modify(newData);
        } else {
            /* LN is not cached, create new LN for logging. */
            ln = dbType.createUpdatedLN(envImpl, newData);
            /* Make new LN resident. */
            bin.attachNode(index, ln, null /*lnSlotKey*/);
            oldLNMemSize = ln.getMemorySizeIncludedByParent();
        }

        /*
         * Create a new WriteLockInfo or get an existing one for the LSN
         * of the current slot, and set its abortLSN and abortKD fields,
         * if needed, i.e, if it is not the current txn the one who created
         * this LSN. The abortLSN and abortKD fields of the wli will be
         * included in the new logrec.
         */
        WriteLockInfo wli = lockStanding.prepareForUpdate();

        /* Log the new record version and lock its new LSN . */
        logResult = ln.optionalLog(
            envImpl, databaseImpl, (key != null ? key : oldKey),
            oldLsn, lastLoggedSize, locker, wli, repContext);

        /* Return a copy of resulting data, if requested. [#16932] */
        if (returnNewData != null) {
            returnNewData.setData(null);
            ln.setEntry(returnNewData);
        }

        /*
         * Update the parent BIN. Update the key, if changed. [#15704]
         */
        bin.updateEntry(
            index, oldLNMemSize, logResult.newLsn, logResult.newSize, key);

        /* Cache record version for update operation. */
        setCurrentVersion(ln.getVLSNSequence(), logResult.newLsn);

        /*
         * It is desirable to evict the LN in a duplicates DB because it
         * will never be fetched again.  But for deferred-write DBs we
         * should not evict a dirty LN since it may be logged unnecessarily.
         */
        if (databaseImpl.getSortedDuplicates() &&
            !databaseImpl.isDeferredWriteMode() &&
            bin.getTarget(index) != null) {
            bin.evictLN(index);
        }

        trace(Level.FINER, TRACE_MOD, bin, index, oldLsn, logResult.newLsn);

        return OperationStatus.SUCCESS;
    }

    /**
     * Position the cursor at the first or last record of the databaseImpl.
     * It's okay if this record is deleted.
     *
     * The cursor must initially be uninitialized.
     *
     * Returns with the target BIN latched!
     *
     * @return true if a first or last position is found, false if the
     * tree being searched is empty.
     */
    public boolean positionFirstOrLast(boolean first)
        throws DatabaseException {

        assert assertCursorState(
            false /*mustBeInitialized*/, true /*mustNotBeInitialized*/);

        boolean found = false;

        try {
            if (first) {
                bin = databaseImpl.getTree().getFirstNode(cacheMode);
            } else {
                bin = databaseImpl.getTree().getLastNode(cacheMode);
            }

            if (bin != null) {

                TreeWalkerStatsAccumulator treeStatsAccumulator =
                    getTreeStatsAccumulator();

                if (bin.getNEntries() == 0) {

                    /*
                     * An IN was found. Even if it's empty, let Cursor
                     * handle moving to the first non-deleted entry.
                     */
                    found = true;
                    index = -1;
                } else {
                    index = (first ? 0 : (bin.getNEntries() - 1));

                    if (treeStatsAccumulator != null &&
                        !bin.isEntryKnownDeleted(index) &&
                        !bin.isEntryPendingDeleted(index)) {
                        treeStatsAccumulator.incrementLNCount();
                    }

                    /*
                     * Even if the entry is deleted, just leave our
                     * position here and return.
                     */
                    found = true;
                }
            }

            addCursor(bin);
            setInitialized();

            return found;
        } catch (final Throwable e) {
            /* Release latch on error. */
            releaseBIN();
            throw e;
        }
    }

    /**
     * Position this cursor on the slot whose key is the max key less or equal
     * to the given search key.
     *
     * To be more precise, let K1 be search key. The method positions the
     * cursor on the BIN that should contain K1. If the BIN does contain K1,
     * this.index is set to the containing slot. Otherwise, this.index is
     * set to the right-most slot whose key is < K1, or to -1 if K1 is < than
     * all keys in the BIN.
     *
     * The cursor must initially be uninitialized.
     *
     * The method returns with the BIN latched, unless an exception is raised.
     *
     * The method returns an integer that encodes the search outcome: If the
     * FOUND bit is not set, the tree is completely empty (has no BINs). If
     * the FOUND bit is set, the EXACT_KEY bit says whether K1 was found or
     * not and the FOUND_LAST bit says whether the cursor is positioned to the
     * very last slot of the BTree (note that this state can only be counted
     * on as long as the BIN is latched).
     *
     * Even if the search returns an exact result, the record may be deleted.
     * The caller must therefore check  whether the cursor is positioned on a
     * deleted record.
     *
     * This method does not lock the record. The caller is expected to call
     * lockAndGetCurrent to perform locking.
     */
    public int searchRange(
        DatabaseEntry searchKey,
        Comparator<byte[]> comparator) {

        assert assertCursorState(
            false /*mustBeInitialized*/, true /*mustNotBeInitialized*/);

        boolean foundSomething = false;
        boolean foundExactKey = false;
        boolean foundLast = false;
        BINBoundary binBoundary = new BINBoundary();

        try {
            byte[] key = Key.makeKey(searchKey);

            bin = databaseImpl.getTree().search(
                key, Tree.SearchType.NORMAL, binBoundary, cacheMode,
                comparator);

            if (bin != null) {

                foundSomething = true;
                if (bin.isBINDelta() && comparator != null) {

                    /*
                     * We must mutate a BIN delta if a non-null comparator is
                     * used. Otherwise, if we positioned the cursor on the
                     * delta using the non-null comparator, we would not be
                     * able to adjust its position correctly later when the
                     * delta gets mutated for some reason (because at that
                     * later time, the comparator used here would not be
                     * known).
                     */
                    bin.mutateToFullBIN();
                }

                index = bin.findEntry(
                    key, true /*indicateIfExact*/, false/*exact*/, comparator);

                if (bin.isBINDelta() &&
                    (index < 0 ||
                     (index & IN.EXACT_MATCH) == 0 ||
                     binBoundary.isLastBin)) {

                    /*
                     * Note: if binBoundary.isLastBin, we must mutate the BIN
                     * in order to compute the foundLast flag below.
                     */
                    bin.mutateToFullBIN();
                    index = bin.findEntry(key, true, false, comparator);
                }

                if (index >= 0) {
                    if ((index & IN.EXACT_MATCH) != 0) {
                        foundExactKey = true;
                        index &= ~IN.EXACT_MATCH;
                    }

                    foundLast = (binBoundary.isLastBin &&
                                 index == bin.getNEntries() - 1);
                }

                /*
                 * Must call addCursor after mutateToFullBIN() to avoid having
                 * to reposition "this" inside mutateToFullBIN(), which would
                 * be both unnecessary and wrong given that this.index could
                 * have the IN.EXACT_MATCH still on.
                 */
                addCursor(bin);
            }

            setInitialized();

            /* Return a multi-part status value */
            return ((foundSomething ? FOUND : 0) |
                    (foundExactKey ? EXACT_KEY : 0) |
                    (foundLast ? FOUND_LAST : 0));

        } catch (final Throwable e) {
            releaseBIN();
            throw e;
        }
    }

    public boolean searchExact(DatabaseEntry searchKey, LockType lockType) {
        return searchExact(searchKey, lockType, false, false) != null;
    }

    /**
     * Position this cursor on the slot (if any) whose key matches the given
     * search key. If no such slot is found or the slot does not hold a "valid"
     * record, return null. Otherwise, lock the found record with the specified
     * lock type (which may be NONE) and return the LockStanding obj that was
     * created by the locking op. Whether the slot contains a "valid" record or
     * not depends on the slot's KD/PD flags and the lockType and dirtyReadAll
     * parameters. Four cases are considered; they are described in the
     * lockLNAndCheckDeleted() method.
     *
     * The cursor must initially be uninitialized.
     *
     * The method returns with the BIN latched, unless an exception is raised.
     *
     * In all cases, the method registers the cursor with the BIN that contains
     * or should contain the search key.
     *
     * @return the LockStanding for the found record, or null if no record was
     * found.
     */
    public LockStanding searchExact(
        final DatabaseEntry searchKey,
        final LockType lockType,
        final boolean dirtyReadAll,
        final boolean dataRequested) {

        assert assertCursorState(
            false /*mustBeInitialized*/, true /*mustNotBeInitialized*/);

        LockStanding lockStanding = null;

        try {
            byte[] key = Key.makeKey(searchKey);

            bin = databaseImpl.getTree().search(key, cacheMode);

            if (bin != null) {

                index = bin.findEntry(key, false, true /*exact*/);

                if (index < 0 && bin.isBINDelta()) {

                    if (bin.mayHaveKeyInFullBin(key)) {
                        bin.mutateToFullBIN();
                        index = bin.findEntry(key, false, true /*exact*/);
                    }
                }

                addCursor(bin);

                if (index >= 0) {
                    lockStanding = lockLNAndCheckDeleted(
                        lockType, dirtyReadAll, dataRequested);
                }
            }

            setInitialized();
            return lockStanding;

        } catch (final Throwable e) {
            /* Release latch on error. */
            releaseBIN();
            throw e;
        }
    }

    /**
     * Lock and copy current record into the key and data DatabaseEntry.
     * When calling this method, this.bin should not be latched already.
     * On return, this.bin is unlatched.
     */
    public OperationStatus lockAndGetCurrent(
        DatabaseEntry foundKey,
        DatabaseEntry foundData,
        final LockType lockType)
        throws DatabaseException {

        return lockAndGetCurrent(
            foundKey, foundData, lockType, false, false, true);
    }

    /**
     * Let S be the slot where this cursor is currently positioned on. If S
     * does not hold a "valid" record, return KEYEMPTY. Otherwise, lock the
     * record in S with the specified lock type(which may be NONE), copy its
     * key and data into the key and data DatabaseEntries, and return SUCCESS.
     * Whether the slot contains a "valid" record or not depends on the slot's
     * KD/PD flags and the lockType and dirtyReadAll parameters. Four cases are
     * considered; they are described in the lockLNAndCheckDeleted() method.
     *
     * On entry, the isLatched param says whether this.bin is latched or not.
     * On return, this.bin is unlatched if the unlatch param is true or an
     * exception is thrown.
     */
    public OperationStatus lockAndGetCurrent(
        DatabaseEntry foundKey,
        DatabaseEntry foundData,
        final LockType lockType,
        final boolean dirtyReadAll,
        final boolean isLatched,
        final boolean unlatch)
        throws DatabaseException {

        /* Used in the finally to indicate whether exception was raised. */
        boolean success = false;

        try {
            assert assertCursorState(
                true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);

            assert checkAlreadyLatched(isLatched) : dumpToString(true);

            if (!isLatched) {
                latchBIN();
            }

            assert(bin.getCursorSet().contains(this));

            TreeWalkerStatsAccumulator treeStatsAccumulator =
                getTreeStatsAccumulator();

            /*
             * If we encounter a deleted slot, opportunistically add the BIN
             * to the compressor queue.
             */
            if (index >= 0 && index < bin.getNEntries() &&
                (bin.isEntryKnownDeleted(index) ||
                 bin.isEntryPendingDeleted(index))) {
                bin.queueSlotDeletion();
            }

            /*
             * Check the KD flag in the BIN slot and make sure this isn't an
             * empty BIN. The BIN could be empty by virtue of the compressor
             * running the size of this BIN to 0 but not having yet deleted
             * it from the tree.
             *
             * The index may be negative if we're at an intermediate stage in
             * an higher level operation (e.g., the starting search for a range
             * scan op), and we expect a higher level method to do a next or
             * prev operation after this returns KEYEMPTY. [#11700]
             */
            if (index < 0 ||
                index >= bin.getNEntries() ||
                bin.isEntryKnownDeleted(index)) {
                /* Node is no longer present. */
                if (treeStatsAccumulator != null) {
                    treeStatsAccumulator.incrementDeletedLNCount();
                }

                success = true;
                return OperationStatus.KEYEMPTY;
            }

            assert TestHookExecute.doHookIfSet(testHook);

            final boolean dataRequested =
                (foundData != null &&
                (!foundData.getPartial() ||
                 foundData.getPartialLength() != 0));

            if (lockLNAndCheckDeleted(
                lockType, dirtyReadAll, dataRequested) == null) {
                if (treeStatsAccumulator != null) {
                    treeStatsAccumulator.incrementDeletedLNCount();
                }
                success = true;
                return OperationStatus.KEYEMPTY;
            }

            getCurrent(foundKey, foundData);

            success = true;
            return OperationStatus.SUCCESS;

        } finally {
            if (unlatch || !success) {
                releaseBIN();
            }
        }
    }

    /**
     * Let S be the slot where this cursor is currently positioned on. The
     * method locks S (i.e. its LSN), and depending on S's KD/PD flags, it
     * returns either null or the LockStanding obj that was created by the
     * locking op. The following 4 cases are considered:
     *
     * 1. If S is not KD/PD, return the LockStanding obj. In this case, we
     * know that S holds a valid (non-deleted) record.
     *
     * 2. If S is KD/PD and the lock type is not NONE, return null. In this
     * case, we know that the record that used to be in S has been definitely
     * deleted.
     *
     * 3. If S is KD/PD, the lock kind is NONE, and dirtyReadAll is false,
     * return null. This case corresponds to the READ_UNCOMMITTED LockMode.
     * The record in S has been deleted, but the deleting txn may be active
     * still, and if it aborts later, the record will be restored. To avoid a
     * potentially blocking lock, in READ_UNCOMMITTED mode we consider the
     * record to be non-existing and return null.
     *
     * 4. If S is KD/PD, the lock kind is NONE, and dirtyReadAll is true, lock
     * the record in READ mode. This case corresponds to the
     * READ_UNCOMMITTED_ALL LockMode, which requires that we do not skip
     * "provisionally deleted" records. There are two sub-cases:
     *
     *  4a. If dataRequested is false, we wait until the deleting txn finishes.
     *      In this case the READ lock is blocking. If after the lock is
     *      granted S is still KD/PD, release the lock and return null.
     *      Otherwise, release the lock and return the LockStanding obj.
     *
     *  4b. If dataRequested is true, then we check whether the deleting txn is
     *      still open by requested a non-blocking READ lock. If the lock is
     *      granted then the deleting txn is closed or this cursor's locker is
     *      the deleter, and we proceed as if the READ lock was granted in 4a.
     *      If the lock is denied then the deleting txn is still open, and we
     *      return the LockStanding obj so that the record is not skipped.
     *
     * The BIN must be latched on entry and is latched on exit.
     *
     * @param dirtyReadAll is true if using LockMode.READ_UNCOMMITTED_ALL.
     *
     * @param dataRequested is true if the read operation should return the
     * record data, meaning that a blocking lock must be used for dirtyReadAll.
     * Is ignored if dirtyReadAll is false. Is always false for a dup DB,
     * since data is never requested for dup DB ops at the CursorImpl level.
     */
    private LockStanding lockLNAndCheckDeleted(
        final LockType lockType,
        final boolean dirtyReadAll,
        final boolean dataRequested) {

        assert !(dirtyReadAll && lockType != LockType.NONE);
        assert !(dataRequested && databaseImpl.getSortedDuplicates());

        LockStanding standing = lockLN(lockType);

        if (standing.recordExists()) {
            return standing;
        }

        /* The slot is KD/PD. */

        if (lockType != LockType.NONE) {
            revertLock(standing);
            /*
             * The deletion was committed by another locker, or has been
             * performed by this locker.
             */
            return null;
        }

        /* We're using dirty-read.  The lockLN above did not actually lock. */

        if (!dirtyReadAll) {
            /* READ_UNCOMMITTED -- skip deleted records without locking. */
            return null;
        }

        /*
         * READ_UNCOMMITTED_ALL -- get a read lock. Whether we can request a
         * no-wait or a blocking lock depends on the dataRequested parameter.
         *
         * Although there is some redundant processing in the sense that lockLN
         * is called more than once (above and below), this is not considered a
         * performance issue because accessing deleted records is normally
         * infrequent. Deleted slots are normally compressed away quickly.
         */
        standing = lockLN(
            LockType.READ, false /*allowUncontended*/,
            !dataRequested /*noWait*/);

        if (standing.lockResult.getLockGrant() == LockGrantType.DENIED) {

            /*
             * The no-wait lock request was denied, which means the data is not
             * needed and the deleting transaction is still open. The deleted
             * record should not be skipped in this case, according to the
             * definition of READ_UNCOMMITTED_ALL.
             */
            assert !standing.recordExists();
            return standing;
        }

        /* We have acquired a temporary read lock. */
        revertLock(standing);

        if (standing.recordExists()) {
            /* Another txn aborted the deletion while we waited. */
            return standing;
        }

        /*
         * The deletion was committed by another locker, or has been performed
         * by this locker.
         */
        return null;
    }

    /**
     * Copy current record into the key and data DatabaseEntry.
     */
    public void getCurrent(
        DatabaseEntry foundKey,
        DatabaseEntry foundData) {

        assert(bin.isLatchExclusiveOwner());
        assert(index >= 0 && index < bin.getNEntries());
        assert(!bin.isEntryKnownDeleted(index));

        /*
         * We don't need to fetch the LN if the user has not requested that we
         * return the data, or if we know for sure that the LN is empty.
         */
        final boolean isEmptyLN = databaseImpl.isLNImmediatelyObsolete();

        final boolean dataRequested =
            (foundData != null &&
             (!foundData.getPartial() || foundData.getPartialLength() != 0));

        final LN ln = (!isEmptyLN && dataRequested ?
                       bin.fetchLN(index, cacheMode) :
                       null);

        /* Return the data. */
        if (dataRequested) {
            assert(ln != null || isEmptyLN);

            byte[] lnData =
                (ln != null ? ln.getData() : LogUtils.ZERO_LENGTH_BYTE_ARRAY);

            LN.setEntry(foundData, lnData);
        }

        /* Return the key */
        if (foundKey != null) {
            LN.setEntry(foundKey, bin.getKey(index));
        }

        /* Cache record version for fetch operation. */
        final long vlsn = (ln != null ?
                           ln.getVLSNSequence() :
                           bin.getVLSN(index, false /*allowFetch*/, cacheMode));

        setCurrentVersion(vlsn, bin.getLsn(index));
    }

    public LN getCurrentLN(final boolean isLatched, final boolean unlatch)
        throws DatabaseException {

        /* Used in the finally to indicate whether exception was raised. */
        boolean success = false;

        try {
            assert assertCursorState(
                true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);
            assert checkAlreadyLatched(isLatched) : dumpToString(true);

            if (!isLatched) {
                latchBIN();
            }

            assert(bin.getCursorSet().contains(this));

            LN ln = bin.fetchLN(index, cacheMode);

            success = true;
            return ln;
        } finally {
            if (unlatch || !success) {
                releaseBIN();
            }
        }
    }

    /**
     * Retrieve the current LN, return with the target bin unlatched.
     */
    public LN lockAndGetCurrentLN(final LockType lockType)
        throws DatabaseException {

        return lockAndGetCurrentLN(lockType, false, true);
    }

    /**
     * Retrieve the current LN. On entry, the isLatched param says whether
     * this.bin is latched or not. On return, this.bin is unlatched if the
     * unlatch param is true or an exception is thrown.
     */
    public LN lockAndGetCurrentLN(final LockType lockType,
                                  final boolean isLatched,
                                  final boolean unlatch)
        throws DatabaseException {

        /* Used in the finally to indicate whether exception was raised. */
        boolean success = false;

        try {
            assert assertCursorState(
                true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);
            assert checkAlreadyLatched(isLatched) : dumpToString(true);

            if (!isLatched) {
                latchBIN();
            }

            assert(bin.getCursorSet().contains(this));

            LockStanding lockStanding = lockLN(lockType);
            if (!lockStanding.recordExists()) {
                revertLock(lockStanding);
                success = true;
                return null;
            }

            LN ln = bin.fetchLN(index, cacheMode);

            success = true;
            return ln;
        } finally {
            if (unlatch || !success) {
                releaseBIN();
            }
        }
    }

    /**
     * Returns the VLSN and LSN for the record at the current position.  Must
     * be called when the cursor is positioned on a record.
     *
     * If this method is called on a secondary cursor, the version of the
     * associated primary record is returned.  In that case, the allowFetch
     * parameter is ignored, and the version is available only if the primary
     * record was retrieved (see setSecondaryCurrentVersion).
     *
     * @param allowFetch is true to fetch the LN to get the VLSN, or false to
     * return -1 for the VLSN if both the LN and VLSN are not cached.
     *
     * @throws IllegalStateException if the cursor is closed or uninitialized,
     * or this is a secondary cursor and the version is not cached.
     */
    public RecordVersion getCurrentVersion(boolean allowFetch) {

        /* Ensure cursor is open and initialized. */
        checkCursorState(
            true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);

        /*
         * For a secondary cursor, the cached version is all we have.
         * See setSecondaryCurrentVersion.
         */
        if (isSecondaryCursor) {
            if (currentRecordVersion == null) {
                throw new IllegalStateException(
                    "Record version is available via a SecondaryCursor only " +
                    "if the associated primary record was retrieved.");
            }
            return currentRecordVersion;
        }

        /*
         * Use cached version if available.  Do not use cached version if it
         * does not contain a VLSN, and VLSNs are preserved, and fetching is
         * allowed; instead, try to fetch it below.
         */
        if (currentRecordVersion != null) {
            if ((currentRecordVersion.getVLSN() !=
                 VLSN.NULL_VLSN.getSequence()) ||
                !allowFetch ||
                !databaseImpl.getEnv().getPreserveVLSN()) {
                return currentRecordVersion;
            }
        }

        /* Get the VLSN from the BIN, create the version and cache it. */
        latchBIN();
        try {
            setCurrentVersion(
                bin.getVLSN(index, allowFetch, cacheMode), bin.getLsn(index));
        } finally {
            releaseBIN();
        }
        return currentRecordVersion;
    }

    private void setCurrentVersion(long vlsn, long lsn) {
        if (isSecondaryCursor) {
            /* NOP.  Version is set by setSecondaryCurrentVersion. */
            return;
        }
        currentRecordVersion = new RecordVersion(vlsn, lsn);
    }

    /**
     * Returns the cached version without attempting to construct it.
     */
    public RecordVersion getCachedRecordVersion() {
        return currentRecordVersion;
    }

    /**
     * When the primary record is read during a secondary operation, this
     * method is called to copy the primary version here. Since secondary
     * records do not have a version of their own, the secondary cursor
     * returns the version of the primary record.
     *
     * @param primaryVersion the version retrieved from the primary cursor
     * using getCachedRecordVersion.
     */
    public void setSecondaryCurrentVersion(RecordVersion primaryVersion) {
        assert isSecondaryCursor;
        currentRecordVersion = primaryVersion;
    }

    /**
     * Advance a cursor.  Used so that verify can advance a cursor even in the
     * face of an exception [12932].
     * @param key on return contains the key if available, or null.
     * @param data on return contains the data if available, or null.
     */
    public boolean advanceCursor(DatabaseEntry key, DatabaseEntry data) {

        BIN oldBin = bin;
        int oldIndex = index;

        key.setData(null);
        data.setData(null);

        try {
            getNext(
                key, data, LockType.NONE, false /*dirtyReadAll*/,
                true /*forward*/, false /*isLatched*/,
                null /*rangeConstraint*/);
        } catch (DatabaseException ignored) {
            /* Klockwork - ok */
        }

        /*
         * If the position changed, regardless of an exception, then we believe
         * that we have advanced the cursor.
         */
        if (bin != oldBin || index != oldIndex) {

            /*
             * Return the key and data from the BIN entries, if we were not
             * able to read it above.
             */
            if (key.getData() == null && bin != null && index > 0) {
                LN.setEntry(key, bin.getKey(index));
            }
            return true;
        } else {
            return false;
        }
    }

    /**
     * Move the cursor forward and return the next "valid" record. Whether a
     * slot contains a "valid" record or not depends on the slot's KD/PD flags
     * and the lockType and dirtyReadAll parameters. Four cases are considered;
     * they are described in the lockLNAndCheckDeleted() method.
     *
     * This will cross BIN boundaries. On return, no latches are held. If no
     * exceptions, the cursor is registered with its new location.
     *
     * @param foundKey DatabaseEntry to use for returning key
     *
     * @param foundData DatabaseEntry to use for returning data
     *
     * @param forward if true, move forward, else move backwards
     *
     * @param isLatched if true, the bin that we're on is already
     * latched.
     *
     * @param rangeConstraint if non-null, is called to determine whether a key
     * is out of range.
     *
     * @return the status.
     */
    public OperationStatus getNext(
        DatabaseEntry foundKey,
        DatabaseEntry foundData,
        LockType lockType,
        boolean dirtyReadAll,
        boolean forward,
        boolean isLatched,
        RangeConstraint rangeConstraint)
        throws DatabaseException {

        assert assertCursorState(
            true /*mustBeInitialized*/, false /*mustNotBeInitialized*/);

        assert checkAlreadyLatched(isLatched) : dumpToString(true);

        OperationStatus result = OperationStatus.NOTFOUND;
        BIN anchorBIN = null;

        try {
            while (bin != null) {

                assert checkAlreadyLatched(isLatched) : dumpToString(true);

                if (!isLatched) {
                    latchBIN();
                    isLatched = true;
                }

                if (DEBUG) {
                    verifyCursor(bin);
                }

                bin.mutateToFullBIN();

                /* Is there anything left on this BIN? */
                if ((forward && ++index < bin.getNEntries()) ||
                    (!forward && --index > -1)) {

                    if (rangeConstraint != null &&
                        !rangeConstraint.inBounds(bin.getKey(index))) {

                        result = OperationStatus.NOTFOUND;
                        releaseBIN();
                        break;
                    }

                    OperationStatus ret = lockAndGetCurrent(
                        foundKey, foundData, lockType, dirtyReadAll,
                        true /*isLatched*/, false /*unlatch*/);

                    if (LatchSupport.TRACK_LATCHES) {
                        LatchSupport.expectBtreeLatchesHeld(1);
                    }

                    if (ret == OperationStatus.SUCCESS) {
                        incrementLNCount();
                        releaseBIN();
                        result = OperationStatus.SUCCESS;
                        break;
                    } else {
                        continue;
                    }
                } else {
                    /*
                     * Make sure that the current BIN will not be pruned away
                     * if it is or becomes empty after it gets unlatched by
                     * Tree.getNextBin() or Tree.getPrevBin(). The operation
                     * of these Tree methods relies on the current BIN not
                     * getting pruned.
                     */
                    anchorBIN = bin;
                    anchorBIN.pin();
                    bin.removeCursor(this);
                    bin = null;

                    final Tree tree = databaseImpl.getTree();

                    /* SR #12736 Try to prune away oldBin */
                    assert TestHookExecute.doHookIfSet(testHook);

                    if (forward) {
                        bin = tree.getNextBin(anchorBIN, cacheMode);
                        index = -1;
                    } else {
                        bin = tree.getPrevBin(anchorBIN, cacheMode);
                        if (bin != null) {
                            index = bin.getNEntries();
                        }
                    }
                    isLatched = true;

                    if (bin == null) {
                        if (LatchSupport.TRACK_LATCHES) {
                            LatchSupport.expectBtreeLatchesHeld(0);
                        }
                        result = OperationStatus.NOTFOUND;
                        break;
                    } else {
                        if (LatchSupport.TRACK_LATCHES) {
                            LatchSupport.expectBtreeLatchesHeld(1);
                        }

                        addCursor();
                        anchorBIN.unpin();
                        anchorBIN = null;
                    }
                }
            }
        } finally {
            if (anchorBIN != null) {
                anchorBIN.unpin();
            }
        }

        if (LatchSupport.TRACK_LATCHES) {
            LatchSupport.expectBtreeLatchesHeld(0);
        }

        return result;
    }

    /**
     * Used to detect phantoms during "get next" operations with serializable
     * isolation.  If this method returns true, the caller should restart the
     * operation from the prior position.
     *
     * Something may have been added to the original cursor (cursorImpl) while
     * we were getting the next BIN.  cursorImpl would have been adjusted
     * properly but we would have skipped a BIN in the process.  This can
     * happen when all INs are unlatched in Tree.getNextBin.  It can also
     * happen without a split, simply due to inserted entries in the previous
     * BIN.
     *
     * @return true if an unaccounted for insertion happened.
     *
     * TODO:
     * Unfortunately, this method doesn't cover all cases where a phantom may
     * have been inserted.  Another case is described below.
     *
     *                        IN-0
     *                        ----------------------
     *                        |    | 50 | 100 |    |
     *                        ----------------------
     *                        /    /         \     \
     *                            /           \
     *                       /----             ----\
     *        IN-1          /                       \       IN-2
     *        ----------------                   ----------------
     *        | 60 | 70 | 80 |                   |    |    |    |
     *        ----------------                   ----------------
     *         /      |      \                     /
     *        /       |       \                   /
     *                   ----------------     -----------------
     *                   | 81 | 83 | 85 |     | 110 |    |    |
     *                   ----------------     -----------------
     *                   BIN-3                BIN-4
     *
     * Initially, the tree looks as above and a cursor (C) is located on the
     * last slot of BIN-3. For simplicity, assume no duplicates and no
     * serializable isolation. Also assume that C is a sticky cursor.
     *
     * 1. Thread 1 calls C.getNext(), which calls retrieveNextAllowPhantoms(),
     *    which duplicates C's cursorImpl, and calls dup.getNext().
     *
     *    dup.getNext() latches BIN-3, sets dup.binToBeRemoved to BIN-3 and
     *    then calls Tree.getNextBin(BIN-3).
     *
     *    Tree.getNextBin(BIN-3) does the following:
     *    - sets searchKey to 85
     *    - calls Tree.getParentINForChildIN(BIN-3).
     *    - Tree.getParentINForChildIN(BIN-3) unlatches BIN-3 and searches for
     *       BIN-3 parent, thus reaching IN-1.
     *    - IN-1.findEntry(85) sets "index" to 2,
     *    - "index" is incremented,
     *    - "moreEntriesThisIn" is set to false,
     *    - "next" is set to IN-1, .
     *    - Tree.getParentINForChildIN(IN-1) is called and unlatches IN-1.
     *
     *    Assume at this point thread 1 looses the cpu.
     *
     * 2. Thread 2 inserts keys 90 and 95, causing a split of both BIN-3 and
     *    IN-1. So the tree now looks like this:
     *
     *                       IN-0
     *                       ---------------------------
     *                       |    | 50 | 80 | 100 |    |
     *                       ---------------------------
     *                       /    /      |      \      \
     *                           /       |       \
     *                 /---------        |        ----------\
     *       IN-1     /                  |                   \       IN-2
     *               /              IN-5 |                    \
     *       -----------            -----------               ----------------
     *       | 60 | 70 |            | 80 | 90 |               |    |    |    |
     *       -----------            -----------               ----------------
     *        /      |              /         \                /
     *       /       |             /           \              /
     *                     ----------------  -----------   -----------------
     *                     | 81 | 83 | 85 |  | 90 | 95 |   | 110 |    |    |
     *                     ----------------  -----------   -----------------
     *                     BIN-3             BIN-6         BIN-4
     *
     *
     *    Notice that C.cursorImpl still points to the last slot of BIN-3.
     *
     * 3. Thread 1 resumes:
     *
     *    - Tree.getParentINForChildIN(IN-1) reaches IN-0.
     *    - IN-0.findEntry(85) sets "index" to 2,
     *    - "index" is incremented,
     *    - "nextIN" is set to IN-2, which is latched.
     *    - Tree.searchSubTree(IN-2, LEFT) is called, and returns BIN-4.
     *    - BIN-4 is the result of Tree.getNextBin(BIN-3), i.e., BIN-6 was
     *      skipped
     *
     *      Now we are back in dup.getNext():
     *    - dup.bin is set to BIN-4, dup.index to -1, and dup is added to BIN-4
     *    - the while loop repeats, dup.index is set to 0, the 1st slot of
     *      BIN-4 is locked, and dup.getNext() returns SUCCESS.
     *
     *      Now we are back in C.retrieveNextAllowPhantoms():
     *    - C.checkForInsertion() is called
     *    - C.cursorImpl and dup are on different BINs, but the condition:
     *      origBIN.getNEntries() - 1 > origCursor.getIndex()
     *      is false, so C.checkForInsertion() returns false.
     *
     * The end result is that BIN-6 has been missed. This is not be a "bug" for
     * non-serializable isolation, but the above scenario applies to
     * serializable isolation as well, and in that case, BIN-6 should really
     * not be missed. This could be solved by re-implementing
     * Tree.getNext/PrevBIN() do a more "logical" kind of search.
     */
    public boolean checkForInsertion(
        final GetMode getMode,
        final CursorImpl dupCursor) {

        final CursorImpl origCursor = this;
        boolean forward = getMode.isForward();
        boolean ret = false;

        if (origCursor.bin != dupCursor.bin) {

            /*
             * We jumped to the next BIN during getNext().
             *
             * Be sure to operate on the BIN returned by latchBIN, not a cached
             * var [#21121].
             *
             * Note that a cursor BIN can change after the check above, but
             * that's not relevant; what we're trying to detect are BIN changes
             * during the operation that has already completed.
             *
             * Note that we can call isPendingDeleted without locking.  If we
             * see a non-committed deleted entry, we'll just iterate around in
             * the caller. So a false positive is ok.
             */
            origCursor.latchBIN();
            final BIN origBIN = origCursor.bin;

            origBIN.mutateToFullBIN();

            try {
                if (forward) {
                    if (origBIN.getNEntries() - 1 > origCursor.getIndex()) {

                        /*
                         * We were adjusted to something other than the
                         * last entry so some insertion happened.
                         */
                        for (int i = origCursor.getIndex() + 1;
                             i < origBIN.getNEntries();
                             i++) {
                            if (!origBIN.isEntryKnownDeleted(i) &&
                                !origBIN.isEntryPendingDeleted(i)) {
                                /* See comment above about locking. */
                                ret = true;
                                break;
                            }
                        }
                    }
                } else {
                    if (origCursor.getIndex() > 0) {

                        /*
                         * We were adjusted to something other than the
                         * first entry so some insertion happened.
                         */
                        for (int i = 0; i < origCursor.getIndex(); i++) {
                            if (!origBIN.isEntryKnownDeleted(i) &&
                                !origBIN.isEntryPendingDeleted(i)) {
                                /* See comment above about locking. */
                                ret = true;
                                break;
                            }
                        }
                    }
                }
            } finally {
                origCursor.releaseBIN();
            }
            return ret;
        }
        return false;
    }

    /**
     * Skips over entries until a boundary condition is satisfied, either
     * because maxCount is reached or RangeConstraint.inBounds returns false.
     *
     * If a maxCount is passed, this allows advancing the cursor quickly by N
     * entries.  If a rangeConstraint is passed, this allows returning the
     * entry count after advancing until the predicate returns false, e.g., the
     * number of entries in a key range.  In either case, the number of entries
     * advanced is returned.
     *
     * Optimized to scan using level two of the tree when possible, to avoid
     * calling getNextBin/getPrevBin for every BIN of the database.  All BINs
     * beneath a level two IN can be skipped quickly, with the level two parent
     * IN latched, when all of its children BINs are resident and can be
     * latched without waiting.  When a child BIN is not resident or latching
     * waits, we revert to the getNextBin/getPrevBin approach, to avoid keeping
     * the parent IN latched for long time periods.
     *
     * Although this method positions the cursor on the last non-deleted entry
     * seen (before the boundary condition is satisfied), because it does not
     * lock the LN it is possible that it is deleted by another thread after
     * the BIN is unlatched.
     *
     * @param forward is true to skip forward, false to skip backward.
     *
     * @param maxCount is the maximum number of non-deleted entries to skip,
     * and may be LTE zero if no maximum is enforced.
     *
     * @param rangeConstraint is a predicate that returns false at a position
     * where advancement should stop, or null if no predicate is enforced.
     *
     * @return the number of non-deleted entries that were skipped.
     */
    public long skip(
        boolean forward,
        long maxCount,
        RangeConstraint rangeConstraint) {

        final CursorImpl c = cloneCursor(true /*samePosition*/);

        try {
            return c.skipInternal(forward, maxCount, rangeConstraint, this);
        } catch (final Throwable e) {
            /*
             * Get more info on dbsim duplicate.conf failure when c.close below
             * throws because the BIN latch is already held.  It should have
             * been released by skipInternal and therefore an unexpected
             * exception must have been throw and the error handling must be
             * incorrect.
             */
            e.printStackTrace(System.out);
            throw e;
        } finally {
            c.close();
        }
    }


    /**
     * Use this cursor to reference the current BIN in the traversal, to
     * prevent the current BIN from being compressed away.  But set the given
     * finalPositionCursor (the 'user' cursor) position only at non-deleted
     * entries, since it should be positioned on a valid entry when this method
     * returns.
     */
    private long skipInternal(
        boolean forward,
        long maxCount,
        RangeConstraint rangeConstraint,
        CursorImpl finalPositionCursor) {

        /* Start with the entry at the cursor position. */
        final Tree tree = databaseImpl.getTree();

        latchBIN();

        IN parent = null;
        BIN prevBin = null;
        BIN curBin = bin;
        int curIndex = getIndex();
        long count = 0;
        boolean success = false;

        try {
            while (true) {
                curBin.mutateToFullBIN();

                /* Skip entries in the current BIN. */
                count = skipEntries(
                    forward, maxCount, rangeConstraint, finalPositionCursor,
                    curBin, curIndex, count);

                if (count < 0) {
                    curBin.releaseLatch();
                    success = true;
                    return (- count);
                }

                /*
                 * Get the parent IN at level two. The BIN is unlatched by
                 * getParentINForChildIN.  Before releasing the BIN latch, get
                 * the search key for the last entry.
                 */
                final byte[] idKey =
                    (curBin.getNEntries() == 0 ?
                     curBin.getIdentifierKey() :
                     (forward ?
                      curBin.getKey(curBin.getNEntries() - 1) :
                      curBin.getKey(0)));

                final SearchResult result = tree.getParentINForChildIN(
                    curBin, false, /*useTargetLevel*/
                    true, /*doFetch*/ CacheMode.DEFAULT);

                parent = result.parent;

                if (!result.exactParentFound) {
                    throw EnvironmentFailureException.unexpectedState(
                        "Cannot get parent of BIN id=" +
                        curBin.getNodeId() + " key=" +
                        Arrays.toString(idKey));
                }

                /*
                 * Find and latch previous child BIN by matching idKey rather
                 * than using result.index, as in Tree.getNextIN (see comments
                 * there).
                 */
                int parentIndex = parent.findEntry(idKey, false, false);

                curBin = (BIN) parent.fetchIN(parentIndex);
                curBin.latch();

                if (forward ?
                    (parentIndex < parent.getNEntries() - 1) :
                    (parentIndex > 0)) {

                    /*
                     * There are more entries in the parent. Skip entries for
                     * child BINs that are resident and can be latched no-wait.
                     */
                    final int incr = forward ? 1 : (-1);

                    for (parentIndex += incr;; parentIndex += incr) {

                        prevBin = curBin;
                        curBin = null;

                        /* Break is no more entries in parent. */
                        if ((forward ?
                             parentIndex >= parent.getNEntries() :
                             parentIndex < 0)) {
                            parent.releaseLatch();
                            break;
                        }

                        /*
                         * Latch next child BIN, if cached and unlatched.
                         *
                         * Note that although 2 BINs are latched here, this
                         * can't cause deadlocks because the 2nd latch is
                         * no-wait.
                         */
                        curBin = (BIN) parent.getTarget(parentIndex);

                        if (curBin == null ||
                            !curBin.latchNoWait(CacheMode.DEFAULT)) {
                            parent.releaseLatch();
                            break;
                        }

                        /* Unlatch the prev BIN */
                        prevBin.releaseLatch();
                        prevBin = null;

                        /* Position at new BIN to prevent compression. */
                        setPosition(curBin, -1);

                        curBin.mutateToFullBIN();

                        /* Skip entries in new child BIN. */
                        count = skipEntries(
                            forward, maxCount, rangeConstraint,
                            finalPositionCursor, curBin,
                            forward ? (-1) : curBin.getNEntries(), count);

                        if (count < 0) {
                            parent.releaseLatch();
                            curBin.releaseLatch();
                            success = true;
                            return (- count);
                        }
                    }
                } else {
                    /* No more entries in the parent. */
                    parent.releaseLatch();
                    prevBin = curBin;
                }

                /*
                 * Only the prevBin is still latched here. Move to the next
                 * BIN the "hard" way (i.e., via full tree searches).
                 */
                curBin = forward ?
                    tree.getNextBin(prevBin, CacheMode.DEFAULT) :
                    tree.getPrevBin(prevBin, CacheMode.DEFAULT);

                assert(!prevBin.isLatchOwner());

                if (curBin == null) {
                    success = true;
                    return count;
                }

                prevBin = null;
                curIndex = forward ? (-1) : curBin.getNEntries();

                /* Position at new BIN to prevent compression. */
                setPosition(curBin, -1);
            }
        } finally {
            if (curBin != null && !success) {
                curBin.releaseLatchIfOwner();
            }
            if (prevBin != null && !success) {
                prevBin.releaseLatchIfOwner();
            }
            if (parent != null && !success) {
                parent.releaseLatchIfOwner();
            }

            if (LatchSupport.TRACK_LATCHES) {
                LatchSupport.expectBtreeLatchesHeld(0);
            }
        }
    }

    /**
     * Skip entries in curBin from one past curIndex and onward.  Returns
     * non-negative count if skipping should continue, or negative count if
     * bounds is exceeded.
     */
    private long skipEntries(
        boolean forward,
        long maxCount,
        RangeConstraint rangeConstraint,
        CursorImpl finalPositionCursor,
        BIN curBin,
        int curIndex,
        long count) {

        assert(!curBin.isBINDelta());

        final int incr = forward ? 1 : (-1);

        for (int i = curIndex + incr;; i += incr) {
            if (forward ? (i >= curBin.getNEntries()) : (i < 0)) {
                break;
            }
            if (rangeConstraint != null &&
                !rangeConstraint.inBounds(curBin.getKey(i))) {
                return (- count);
            }
            if (!curBin.isEntryKnownDeleted(i) &&
                !curBin.isEntryPendingDeleted(i)) {
                count += 1;
                finalPositionCursor.setPosition(curBin, i);
                if (maxCount > 0 && count >= maxCount) {
                    return (- count);
                }
            }
        }
        return count;
    }

    /**
     * Returns the stack of ancestor TrackingInfo for the BIN at the cursor, or
     * null if a split occurs and the information returned would be
     * inconsistent.
     *
     * Used by CountEstimator.
     */
    public List<TrackingInfo> getAncestorPath() {

        /*
         * Search for parent of BIN, get TrackingInfo for ancestors.  If the
         * exact parent is not found, a split occurred and null is returned.
         */
        final List<TrackingInfo> trackingList = new ArrayList<TrackingInfo>();

        latchBIN();

        final BIN origBin = bin;
        final Tree tree = databaseImpl.getTree();

        final SearchResult result = tree.getParentINForChildIN(
            origBin, false, /*useTargetLevel*/
            true /*doFetch*/, CacheMode.UNCHANGED, trackingList);

        if (!result.exactParentFound) {
            /* Must have been a split. */
            return null;
        }

        /*
         * The parent was found and is now latched. If the child BIN does not
         * match the cursor's BIN, then a split occurred and null is returned.
         */
        final long binLsn;
        try {
            if (origBin != result.parent.getTarget(result.index) ||
                origBin != bin) {
                /* Must have been a split. */
                return null;
            }

            binLsn = result.parent.getLsn(result.index);
            bin.latch();

        } finally {
            result.parent.releaseLatch();
        }

        /*
         * The child BIN is now latched. Subtract deleted entries from BIN's
         * total entries and adjust the index accordingly.  Add TrackingInfo
         * for child BIN.
         */
        try {
            int binEntries = bin.getNEntries();
            int binIndex = getIndex();

            for (int i = bin.getNEntries() - 1; i >= 0; i -= 1) {

                if (bin.isEntryKnownDeleted(i) ||
                    bin.isEntryPendingDeleted(i)) {

                    binEntries -= 1;
                    if (i < binIndex) {
                        binIndex -= 1;
                    }
                }
            }

            final TrackingInfo info = new TrackingInfo(
                binLsn, bin.getNodeId(), binEntries, binIndex);

            trackingList.add(info);

            return trackingList;

        } finally {
            bin.releaseLatch();
        }
    }

    /**
     * Search for the next key following the given key, and acquire a range
     * insert lock on it.  If there are no more records following the given
     * key, lock the special EOF node for the databaseImpl.
     */
    public void lockNextKeyForInsert(DatabaseEntry key)
        throws DatabaseException {

        DatabaseEntry tempKey = new DatabaseEntry(
            key.getData(), key.getOffset(), key.getSize());

        boolean lockedNextKey = false;
        boolean latched = true;

        try {
            while (true) {

                int searchResult = searchRange(tempKey, null /*comparator*/);

                if ((searchResult & FOUND) != 0 &&
                    (searchResult & FOUND_LAST) == 0) {

                    /*
                     * The search positioned "this" on the BIN that should
                     * contain K1 and this BIN is now latched. If the BIN does
                     * contain K1, this.index points to K1's slot. Otherwise,
                     * this.index points to the right-most slot whose key is
                     * < K1 (or this.index is -1 if K1 is < than all keys in
                     * the BIN). Furthermore, "this" is NOT positioned on the
                     * very last slot of the BTree.
                     *
                     * Call getNext() to advance "this" to the next *valid*
                     * (i.e., not marked deleted) slot and lock that slot in
                     * RANGE_INSERT mode. Normally, getNext() will move the
                     * cursor to the 1st slot with a key K2 > K1. However, it
                     * is possible that K2 <= K1 (see the comments in
                     * Cursor.searchRangeAdvanceAndCheckKey() about how this
                     * can happen. We handle this race condition by restarting
                     * the search.
                     */
                    DatabaseEntry tempData = new DatabaseEntry();
                    tempData.setPartial(0, 0, true);

                    OperationStatus status = getNext(
                        tempKey, tempData, LockType.RANGE_INSERT,
                        false, true, true,
                        null /*rangeConstraint*/);

                    latched = false;

                    if (status == OperationStatus.SUCCESS) {

                        Comparator<byte[]> comparator =
                            databaseImpl.getKeyComparator();

                        int c = Key.compareKeys(tempKey, key, comparator);
                        if (c <= 0) {
                            tempKey.setData(
                                key.getData(), key.getOffset(), key.getSize());
                            continue;
                        }

                        lockedNextKey = true;
                    }
                }

                break;
            }
        } finally {
            if (latched) {
                releaseBIN();
            }
        }

        /* Lock the EOF node if no next key was found. */
        if (!lockedNextKey) {
            lockEof(LockType.RANGE_INSERT);
        }
    }

    /*
     * Locking
     */

    /**
     * Holds the result of a lockLN operation.  A lock may not actually be
     * held (getLockResult may return null) if an uncontended lock is allowed.
     */
    public static class LockStanding {
        private long lsn;
        private boolean nullLsn;
        private boolean deleted;
        private LockResult lockResult;

        public boolean recordExists() {
            return !nullLsn && !deleted;
        }

        public LockResult getLockResult() {
            return lockResult;
        }

        /**
         * Called by update and delete ops, after lockLN() and before logging
         * the LN. It returns a WriteLockInfo that is meant to be passed to
         * the LN logging method, where its info will be included in the LN
         * log entry and also copied into the new WriteLockInfo that will be
         * created for the new LSN.
         *
         * If the locker is not transactional, or the current LSN has not been
         * write-locked before by this locker, a new WriteLockInfo is created
         * here and its abortLsn and abortKD fields are set. (note: even though
         * lockLN() is called before prepareForUpdate(), it may not actually
         * acquire a lock because of the uncontended optimization).
         *
         * Otherwise, a WriteLockInfo exists already. It may have been created
         * by the lockLN() call during the current updating op, or a lockLN()
         * call during an earlier updating op by the same txn. In the later
         * case, the abortLsn and abortKD have been set already and should not
         * be overwriten here.
         */
        public WriteLockInfo prepareForUpdate() {
            final boolean abortKnownDeleted = !recordExists();
            WriteLockInfo wri = (lockResult == null ?
                                 null :
                                 lockResult.getWriteLockInfo());
            if (wri == null) {
                wri = new WriteLockInfo();
                wri.setAbortLsn(lsn);
                wri.setAbortKnownDeleted(abortKnownDeleted);
            } else {
                lockResult.setAbortLsn(lsn, abortKnownDeleted);
            }
            return wri;
        }

        /**
         * Creates WriteLockInfo that is appropriate for a newly inserted slot.
         * The return value is meant to be passed to an LN logging method and
         * copied into the WriteLockInfo for the new LSN.  This method is
         * static because lockLN is never called prior to logging an LN for a
         * newly inserted slot.
         */
        public static WriteLockInfo prepareForInsert() {
            final WriteLockInfo wri = new WriteLockInfo();
            return wri;
        }
    }

    /** Does not allow uncontended locks.  See lockLN(LockType, boolean). */
    public LockStanding lockLN(LockType lockType)
        throws LockConflictException {

        return lockLN(lockType, false /*allowUncontended*/, false /*noWait*/);
    }

    /**
     * Locks the LN at the cursor position.  Attempts to use a non-blocking
     * lock to avoid unlatching/relatching.
     *
     * Retries if necessary, to handle the case where the LSN is changed while
     * the BIN is unlatched.  Because it re-latches the BIN to check the LSN,
     * this serializes access to the LSN for locking, guaranteeing that two
     * lockers cannot obtain conflicting locks on the old and new LSNs.
     *
     * Preconditions: The BIN must be latched.
     *
     * Postconditions: The BIN is latched.
     *
     * LN Locking Rules
     * ----------------
     * The lock ID for an LN is its LSN in the parent BIN slot.  Because the
     * LSN changes when logging the LN, only two methods of locking an LN may
     * be used to support concurrent access:
     *
     * 1. This method may be called to lock the old LSN.  For read operations,
     * that is all that is necessary.  For write operations, the new LSN must
     * be locked after logging it, which is done by all the LN logging methods.
     * Be sure to pass a non-null locker to the LN logging method to lock the
     * LN, unless locking is not desired.
     *
     * 2. A non-blocking lock may be obtained on the old LSN (using
     * Locker.nonBlockingLock rather than this method), as long as the lock is
     * released before the BIN latch is released.  In this case a non-null
     * locker is not passed to the LN logging method; locking the new LSN is
     * unnecessary because no other thread can access the new LSN until the BIN
     * latch is released.
     *
     * The first method is used for all user operations.  The second method is
     * used by the cleaner, when flushing dirty deferred-write LNs, and by
     * certain btree operations.
     *
     * Uncontended Lock Optimization
     * -----------------------------
     * The allowUncontended param is passed as true for update and delete
     * operations as an optimization for the case where no lock on the old LSN
     * is held by any locker.  In this case we don't need to lock the old LSN
     * at all, as long as we log the new LSN before releasing the BIN latch.
     *
     * 1. Latch BIN
     * 2. Determine that no lock/waiter exists for oldLsn
     * 3. Log LN and get newLsn
     * 4. Lock newLsn
     * 5. Update BIN
     * 6. Release BIN latch
     *
     * The oldLsn is never locked, saving operations on the lock table.  The
     * assumption is that another locker will first have to latch the BIN to
     * get oldLsn, before requesting a lock.
     *
     * A potential problem is that the other locker may release the BIN latch
     * before requesting the lock.
     *
     * This Operation        Another Operation
     * --------------        -----------------
     *                       Latch BIN, get oldLsn, release BIN latch
     * Step 1 and 2
     *                       Request lock for oldLsn, granted
     * Step 3 and 4
     *
     * Both operations now believe they have an exclusive lock, but they have
     * locks on different LSNs.
     *
     * However, this problem is handled as long as the other lock is performed
     * using a lockLN method in this class, which will release the lock and
     * retry if the LSN changes while acquiring the lock.  Because it
     * re-latches the BIN to check the LSN, this will serialize access to the
     * LSN for locking, guaranteeing that two conflicting locks cannot be
     * granted on the old and new LSNs.
     *
     * Deferred-Write Locking
     * ----------------------
     * When one of the LN optionalLog methods is called, a deferred-write LN is
     * dirtied but not actually logged.  In order to lock an LN that has been
     * inserted but not yet assigned a true LSN, a transient LSNs is assigned.
     * These LSNs serve to lock the LN but never appear in the log.  See
     * LN.assignTransientLsn.
     *
     * A deferred-write LN is logged when its parent BIN is logged, or when the
     * LN is evicted.  This will replace transient LSNs with durable LSNs.  If
     * a lock is held by a cursor on a deferred-write LN when it is logged, the
     * same lock is acquired on the new LSN by the cursor.  See
     * lockAfterLsnChange.
     *
     * Cleaner Migration Locking
     * -------------------------
     * The cleaner takes a non-blocking read lock on the old LSN before
     * migrating/logging the LN, while holding the BIN latch.  It does not take
     * a lock on the new LSN, since it does not need to retain a lock after
     * releasing the BIN latch.
     *
     * Because a read, not write, lock is taken, other read locks may be held
     * during migration.  After logging, the cleaner calls lockAfterLsnChange
     * to lock the new LSN on behalf of other lockers.
     *
     * For more info on migration locking, see HandleLocker.
     *
     * Historical Notes
     * ----------------
     * In JE 4.1 and earlier, each LN had a node ID that was used for locking,
     * rather than using the LSN.  The node ID changed only if a deleted slot
     * was reused.  The node ID was stored in the LN, requiring that the LN be
     * fetched when locking the LN.  With LSN locking a fetch is not needed.
     *
     * When LN node IDs were used, deferred-write LNs were not assigned an LSN
     * until they were actually logged. Deferred-write LNs were initially
     * assigned a null LSN and transient LSNs were not needed.
     *
     * @param lockType the type of lock requested.
     *
     * @param allowUncontended is true to return immediately (no lock is taken)
     * when no locker holds or waits for the lock.
     *
     * @param noWait is true to perform a no-wait lock request while keeping
     * the BIN latched.  The caller must check the lock result to see whether
     * the lock was granted.
     *
     * @return all information about the lock; see LockStanding.
     *
     * @throws LockConflictException if the lsn is non-null, the lock is
     * contended, and a lock could not be obtained by blocking.
     */
    public LockStanding lockLN(final LockType lockType,
                               final boolean allowUncontended,
                               final boolean noWait)
        throws LockConflictException {

        final LockStanding standing = new LockStanding();
        standing.lsn = bin.getLsn(index);
        standing.deleted = bin.isEntryKnownDeleted(index) ||
                           bin.isEntryPendingDeleted(index);

        /* Ensure that a known-deleted null LSN is not present. */
        if (standing.lsn == DbLsn.NULL_LSN) {
            assert bin.isEntryKnownDeleted(index);
            standing.nullLsn = true;
            return standing;
        }

        /*
         * We can avoid taking a lock if uncontended.  However, we must
         * call preLogWithoutLock to prevent logging on a replica, and as
         * good measure to prepare for undo.
         */
        if (allowUncontended &&
            databaseImpl.getEnv().getTxnManager().getLockManager().
                         isLockUncontended(standing.lsn)) {
            locker.preLogWithoutLock(databaseImpl);
            assert verifyPendingDeleted(lockType);
            return standing;
        }

        /*
         * Try a non-blocking lock first, to avoid unlatching.  If the default
         * is no-wait, use the standard lock method so
         * LockNotAvailableException is thrown; there is no need to try a
         * non-blocking lock twice.
         *
         * Even for dirty-read (LockType.NONE) we must call Locker.lock() since
         * it checks the locker state and may throw LockPreemptedException.
         */
        if (locker.getDefaultNoWait()) {
            try {
                standing.lockResult = locker.lock(
                    standing.lsn, lockType, true /*noWait*/, databaseImpl);
            } catch (LockConflictException e) {

                /*
                 * Release all latches.  Note that we catch
                 * LockConflictException for simplicity but we expect either
                 * LockNotAvailableException or LockNotGrantedException.
                 */
                releaseBIN();
                throw e;
            }
        } else {
            standing.lockResult = locker.nonBlockingLock(
                standing.lsn, lockType, false /*jumpAheadOfWaiters*/,
                databaseImpl);
        }

        if (noWait ||
            standing.lockResult.getLockGrant() != LockGrantType.DENIED) {
            /* Lock was granted whiled latched, no need to check LSN. */
            assert verifyPendingDeleted(lockType);
            return standing;
        }

        /*
         * Unlatch, get a blocking lock, latch, and get the current LSN from
         * the slot.  If the LSN changes while unlatched, revert the lock and
         * repeat.
         */
        while (true) {

            /* Request a blocking lock. */
            releaseBIN();
            standing.lockResult = locker.lock(
                standing.lsn, lockType, false /*noWait*/, databaseImpl);

            /* Check current LSN after locking. */
            latchBIN();
            standing.deleted = bin.isEntryKnownDeleted(index) ||
                               bin.isEntryPendingDeleted(index);
            final long newLsn = bin.getLsn(index);
            if (standing.lsn != newLsn) {
                /* The LSN changed, revert the lock and try again. */
                revertLock(standing);
                standing.lsn = newLsn;
                /* Ensure that a known-deleted null LSN is not present. */
                if (newLsn == DbLsn.NULL_LSN) {
                    assert bin.isEntryKnownDeleted(index);
                    standing.nullLsn = true;
                    return standing;
                }
                continue;
            } else {
                /* If locked correctly, return the result. */
                assert verifyPendingDeleted(lockType);
                return standing;
            }
        }
    }

    /**
     * After logging a deferred-write LN during eviction/checkpoint or a
     * migrated LN during cleaning, for every existing lock on the old LSN held
     * by another locker, we must lock the new LSN on behalf of that locker.
     *
     * This is done while holding the BIN latch so that the new LSN does not
     * change during the locking process.  The BIN must be latched on entry and
     * is left latched by this method.
     *
     * We release the lock on the oldLsn to prevent locks from accumulating
     * over time on a HandleLocker, as the cleaner migrates LNs, because
     * Database handle locks are legitimately very long-lived.  It is important
     * to first acquire all newLsn locks and then release the oldLsn locks.
     * Releasing an oldLsn lock might allow another locker to acquire it, and
     * then acquiring another newLsn lock may encounter a conflict. [#20617]
     *
     * @see com.sleepycat.je.txn.HandleLocker
     * @see #lockLN
     */
    public static void lockAfterLsnChange(DatabaseImpl dbImpl,
                                          long oldLsn,
                                          long newLsn,
                                          Locker excludeLocker) {
        final LockManager lockManager =
            dbImpl.getEnv().getTxnManager().getLockManager();
        final Set<LockInfo> owners = lockManager.getOwners(oldLsn);
        if (owners == null) {
            return;
        }
        /* Acquire newLsn locks. */
        for (LockInfo lockInfo : owners) {
            final Locker locker = lockInfo.getLocker();
            if (locker != excludeLocker) {
                locker.lockAfterLsnChange(oldLsn, newLsn, dbImpl);
            }
        }
        /* Release oldLsn locks. */
        for (LockInfo lockInfo : owners) {
            final Locker locker = lockInfo.getLocker();
            if (locker != excludeLocker &&
                locker.allowReleaseLockAfterLsnChange()) {
                locker.releaseLock(oldLsn);
            }
        }
    }

    /**
     * For debugging. Verify that a BINs cursor set refers to the BIN.
     */
    private void verifyCursor(BIN bin)
        throws DatabaseException {

        if (!bin.getCursorSet().contains(this)) {
            throw new EnvironmentFailureException
                (databaseImpl.getEnv(),
                 EnvironmentFailureReason.UNEXPECTED_STATE,
                 "BIN cursorSet is inconsistent");
        }
    }

    /**
     * Calls checkCursorState and asserts false if an exception is thrown.
     * Otherwise returns true, so it can be called under an assertion.
     */
    private boolean assertCursorState(boolean mustBeInitialized,
                                      boolean mustNotBeInitialized) {
        try {
            checkCursorState(mustBeInitialized, mustNotBeInitialized);
            return true;
        } catch (RuntimeException e) {
            assert false : e.toString() + " " + dumpToString(true);
            return false; // for compiler
        }
    }

    /**
     * Check that the cursor is open and optionally if it is initialized or
     * uninitialized.
     *
     * @throws IllegalStateException via all Cursor methods that call
     * Cursor.checkState (all get and put methods, plus more).
     */
    public void checkCursorState(boolean mustBeInitialized,
                                 boolean mustNotBeInitialized) {
        switch (status) {
        case CURSOR_NOT_INITIALIZED:
            if (mustBeInitialized) {
                throw new IllegalStateException("Cursor not initialized.");
            }
            break;
        case CURSOR_INITIALIZED:
            if (mustNotBeInitialized) {
                throw EnvironmentFailureException.unexpectedState(
                    "Cursor is initialized.");
            }
            if (DEBUG) {
                if (bin != null) {
                    verifyCursor(bin);
                }
            }
            break;
        case CURSOR_CLOSED:
            throw new IllegalStateException("Cursor has been closed.");
        default:
            throw EnvironmentFailureException.unexpectedState(
                "Unknown cursor status: " + status);
        }
    }

    /**
     * Checks that LN deletedness matches KD/PD flag state, at least when the
     * LN is resident.  Should only be called under an assertion.
     */
    private boolean verifyPendingDeleted(LockType lockType) {

        /* Cannot verify deletedness if LN is not locked. */
        if (lockType == LockType.NONE) {
            return true;
        }

        /* Cannot verify deletedness if cursor is not intialized. */
        if (bin == null || index < 0) {
            return true;
        }

        /* Cannot verify deletedness if LN is not resident. */
        final LN ln = (LN) bin.getTarget(index);
        if (ln == null) {
            return true;
        }

        /*
         * If the LN is deleted then KD or PD must be set.  If the LN is not
         * deleted then PD must not be set, but KD may or may not be set since
         * it used for various purposes (see IN.java).
         */
        final boolean kd = bin.isEntryKnownDeleted(index);
        final boolean pd = bin.isEntryPendingDeleted(index);
        final boolean lnDeleted = ln.isDeleted();
        assert ((lnDeleted && (kd || pd)) || (!lnDeleted && !pd)) :
               "Deleted state mismatch LNDeleted = " + lnDeleted +
               " PD = " + pd + " KD = " + kd;
        return true;
    }

    public void revertLock(LockStanding standing)
        throws DatabaseException {

        if (standing.lockResult != null) {
            revertLock(standing.lsn, standing.lockResult);
            standing.lockResult = null;
        }
    }

    /**
     * Return this lock to its prior status. If the lock was just obtained,
     * release it. If it was promoted, demote it.
     */
    private void revertLock(long lsn, LockResult lockResult)
        throws DatabaseException {

        LockGrantType lockStatus = lockResult.getLockGrant();
        if ((lockStatus == LockGrantType.NEW) ||
            (lockStatus == LockGrantType.WAIT_NEW)) {
            locker.releaseLock(lsn);
        } else if ((lockStatus == LockGrantType.PROMOTION) ||
                   (lockStatus == LockGrantType.WAIT_PROMOTION)){
            locker.demoteLock(lsn);
        }
    }

    /**
     * Locks the logical EOF node for the databaseImpl.
     */
    public void lockEof(LockType lockType)
        throws DatabaseException {

        locker.lock(databaseImpl.getEofLsn(), lockType,
                    false /*noWait*/, databaseImpl);
    }

    /**
     * @throws EnvironmentFailureException if the underlying environment is
     * invalid.
     */
    public void checkEnv() {
        databaseImpl.getEnv().checkIfInvalid();
    }

    /**
     * Callback object for traverseDbWithCursor.
     */
    public interface WithCursor {

        /**
         * Called for each record in the databaseImpl.
         * @return true to continue or false to stop the enumeration.
         */
        boolean withCursor(CursorImpl cursor,
                           DatabaseEntry key,
                           DatabaseEntry data)
            throws DatabaseException;
    }

    /**
     * Enumerates all records in a databaseImpl non-transactionally and calls
     * the withCursor method for each record.  Stops the enumeration if the
     * callback returns false.
     *
     * @param db DatabaseImpl to traverse.
     *
     * @param lockType non-null LockType for reading records.
     *
     * @param allowEviction should normally be true to evict when performing
     * multiple operations, but may be false if eviction is disallowed in a
     * particular context.
     *
     * @param withCursor callback object.
     */
    public static void traverseDbWithCursor(
        DatabaseImpl db,
        LockType lockType,
        boolean allowEviction,
        WithCursor withCursor)
        throws DatabaseException {

        DatabaseEntry key = new DatabaseEntry();
        DatabaseEntry data = new DatabaseEntry();
        Locker locker = null;
        CursorImpl cursor = null;

        try {
            EnvironmentImpl envImpl = db.getEnv();

            locker = LockerFactory.getInternalReadOperationLocker(envImpl);

            cursor = new CursorImpl(db, locker);
            cursor.setAllowEviction(allowEviction);

            if (cursor.positionFirstOrLast(true /*first*/)) {

                OperationStatus status = cursor.lockAndGetCurrent(
                    key, data, lockType, false /*dirtyReadAll*/,
                    true /*isLatched*/, true /*unlatch*/);

                boolean done = false;
                while (!done) {

                    /*
                     * lockAndGetCurrent may have returned non-SUCCESS
                     * if the first record is deleted, but we can call getNext
                     * below to move forward.
                     */
                    if (status == OperationStatus.SUCCESS) {
                        if (!withCursor.withCursor(cursor, key, data)) {
                            done = true;
                        }
                    }

                    if (!done) {
                        status = cursor.getNext(
                            key, data, lockType, false /*dirtyReadAll*/,
                            true /*forward*/, false /*isLatched*/,
                            null /*rangeConstraint*/);

                        if (status != OperationStatus.SUCCESS) {
                            done = true;
                        }
                    }
                }
            }
        } finally {
            if (cursor != null) {
                cursor.releaseBIN();
                cursor.close();
            }
            if (locker != null) {
                locker.operationEnd();
            }
        }
    }

    /**
     * Dump the cursor for debugging purposes.  Dump the bin that the cursor
     * refers to if verbose is true.
     */
    public void dump(boolean verbose) {
        System.out.println(dumpToString(verbose));
    }

    /**
     * dump the cursor for debugging purposes.
     */
    public void dump() {
        System.out.println(dumpToString(true));
    }

    /*
     * dumper
     */
    private String statusToString(byte status) {
        switch(status) {
        case CURSOR_NOT_INITIALIZED:
            return "CURSOR_NOT_INITIALIZED";
        case CURSOR_INITIALIZED:
            return "CURSOR_INITIALIZED";
        case CURSOR_CLOSED:
            return "CURSOR_CLOSED";
        default:
            return "UNKNOWN (" + Byte.toString(status) + ")";
        }
    }

    /*
     * dumper
     */
    public String dumpToString(boolean verbose) {
        StringBuilder sb = new StringBuilder();

        sb.append("<Cursor idx=\"").append(index).append("\"");
        sb.append(" status=\"").append(statusToString(status)).append("\"");
        sb.append(">\n");
        if (verbose) {
            sb.append((bin == null) ? "" : bin.dumpString(2, true));
        }
        sb.append("\n</Cursor>");

        return sb.toString();
    }

    /*
     * For unit tests
     */
    public StatGroup getLockStats()
        throws DatabaseException {

        return locker.collectStats();
    }

    /**
     * Send trace messages to the java.util.logger. Don't rely on the logger
     * alone to conditionalize whether we send this message, we don't even want
     * to construct the message if the level is not enabled.
     */
    private void trace(Level level,
                       String changeType,
                       BIN theBin,
                       int lnIndex,
                       long oldLsn,
                       long newLsn) {
        EnvironmentImpl envImpl = databaseImpl.getEnv();
        if (envImpl.getLogger().isLoggable(level)) {
            StringBuilder sb = new StringBuilder();
            sb.append(changeType);
            sb.append(" bin=");
            sb.append(theBin.getNodeId());
            sb.append(" lnIdx=");
            sb.append(lnIndex);
            sb.append(" oldLnLsn=");
            sb.append(DbLsn.getNoFormatString(oldLsn));
            sb.append(" newLnLsn=");
            sb.append(DbLsn.getNoFormatString(newLsn));

            LoggerUtils.logMsg
                (envImpl.getLogger(), envImpl, level, sb.toString());
        }
    }

    /**
     * Send trace messages to the java.util.logger. Don't rely on the logger
     * alone to conditionalize whether we send this message, we don't even want
     * to construct the message if the level is not enabled.
     */
    private void traceInsert(Level level,
                             BIN insertingBin,
                             long lnLsn,
                             int index) {
        EnvironmentImpl envImpl = databaseImpl.getEnv();
        if (envImpl.getLogger().isLoggable(level)) {
            StringBuilder sb = new StringBuilder();
            sb.append(TRACE_INSERT);
            sb.append(" bin=");
            sb.append(insertingBin.getNodeId());
            sb.append(" lnLsn=");
            sb.append(DbLsn.getNoFormatString(lnLsn));
            sb.append(" index=");
            sb.append(index);

            LoggerUtils.logMsg(envImpl.getLogger(), envImpl, level,
                               sb.toString());
        }
    }

    /* For unit testing only. */
    public void setTestHook(TestHook hook) {
        testHook = hook;
    }

    /* Check that the target bin is latched. For use in assertions. */
    private boolean checkAlreadyLatched(boolean isLatched) {
        if (isLatched) {
            if (bin != null) {
                return bin.isLatchExclusiveOwner();
            }
        }
        return true;
    }
}