*            If true then this method tests whether there is one record
     *            being identified and removes it only if it is a primordial
     *            AntiValue.
     * @return <code>true</code> if any records were removed.
     * @throws PersistitException
     */
    boolean raw_removeKeyRangeInternal(Key key1, Key key2, boolean fetchFirst, boolean removeOnlyAntiValue)
            throws PersistitException {
        _persistit.checkClosed();
        _persistit.checkSuspended();

        if (_volume.isReadOnly()) {
            throw new ReadOnlyVolumeException(_volume.toString());
        }
        if (Debug.ENABLED) {
            Debug.suspend();
        }
        boolean treeClaimAcquired = false;
        boolean treeWriterClaimRequired = false;
        boolean result = false;

        boolean deallocationRequired = true; // assume until proven false
        boolean tryQuickDelete = true;

        if (!_ignoreTransactions) {
            _transaction.remove(this, key1, key2);
        }

        try {
            //
            // Retry here to get an exclusive Tree latch in the occasional case
            // where pages are being joined.
            //
            for (;;) {

                checkLevelCache();
                int depth = _cacheDepth; // The depth to which we have
                // populated the level cache.

                try {
                    //
                    // First try for a quick delete from a single data page.
                    //
                    if (tryQuickDelete) {
                        Buffer buffer = null;
                        try {
                            int foundAt1 = search(key1, true) & P_MASK;
                            buffer = _levelCache[0]._buffer;

                            if (foundAt1 > buffer.getKeyBlockStart() && foundAt1 < buffer.getKeyBlockEnd()) {
                                int foundAt2 = buffer.findKey(key2) & P_MASK;
                                if (!buffer.isBeforeLeftEdge(foundAt2) && !buffer.isAfterRightEdge(foundAt2)) {
                                    foundAt2 &= P_MASK;
                                    if (foundAt2 < buffer.getKeyBlockEnd()) {
                                        Debug.$assert0.t(foundAt2 >= foundAt1);

                                        if (removeOnlyAntiValue) {
                                            for (int p = foundAt1; p < foundAt2; p += KEYBLOCK_LENGTH) {
                                                if (!buffer.isPrimordialAntiValue(p)) {
                                                    return false;
                                                }
                                            }
                                        }
                                        if (fetchFirst) {
                                            removeFetchFirst(buffer, foundAt1, buffer, foundAt2);
                                        }

                                        final long timestamp = timestamp();
                                        buffer.writePageOnCheckpoint(timestamp);
                                        _volume.getStructure().harvestLongRecords(buffer, foundAt1, foundAt2);

                                        boolean removed = buffer.removeKeys(foundAt1, foundAt2, _spareKey1);
                                        if (removed) {
                                            _tree.bumpChangeCount();
                                            buffer.setDirtyAtTimestamp(timestamp);
                                        }
                                        result = removed;
                                        break;
                                    }
                                }
                            }
                            // If we didn't meet the criteria for quick delete,
                            // then don't try it again on a RetryException.
                            tryQuickDelete = false;
                        } finally {
                            if (buffer != null) {
                                buffer.releaseTouched();
                                buffer = null;
                            }
                        }
                    }

                    /*
                     * This deletion is more complicated and involves an index
                     * search. The tree must be latched.
                     */
                    if (!treeClaimAcquired) {
                        if (!_treeHolder.claim(treeWriterClaimRequired)) {
                            Debug.$assert0.t(false);
                            throw new InUseException("Thread " + Thread.currentThread().getName()
                                    + " failed to get writer claim on " + _tree);
                        }
                        treeClaimAcquired = true;
                    }
                    //
                    // Need to redo this check now that we have a
                    // claim on the Tree.
                    //
                    checkLevelCache();

                    long pageAddr1 = _tree.getRootPageAddr();
                    long pageAddr2 = pageAddr1;

                    for (int level = _cacheDepth; --level >= 0;) {
                        LevelCache lc = _levelCache[level];
                        lc.initRemoveFields();
                        depth = level;

                        int foundAt1 = searchLevel(key1, true, pageAddr1, level, true);
                        int foundAt2 = -1;

                        //
                        // Note: this buffer now has a writer claim on it.
                        //
                        Buffer buffer = lc._buffer;
                        lc._flags |= LEFT_CLAIMED;

                        lc._leftBuffer = buffer;
                        lc._leftFoundAt = foundAt1;
                        boolean samePage = pageAddr2 == pageAddr1;

                        if (samePage) {
                            foundAt2 = buffer.findKey(key2);

                            if (!buffer.isAfterRightEdge(foundAt2)) {
                                lc._rightBuffer = buffer;
                                lc._rightFoundAt = foundAt2;
                            } else {
                                pageAddr2 = buffer.getRightSibling();
                                samePage = false;
                            }
                        }
                        if (!samePage) {
                            //
                            // Since we are spanning pages we need an
                            // exclusive claim on the tree to prevent
                            // an insertion from propagating upward through
                            // the deletion range.
                            //
                            if (!treeWriterClaimRequired) {
                                treeWriterClaimRequired = true;
                                if (!_treeHolder.upgradeClaim()) {
                                    throw RetryException.SINGLE;
                                }
                            }

                            foundAt2 = searchLevel(key2, false, pageAddr2, level, true);

                            buffer = lc._buffer;
                            lc._flags |= RIGHT_CLAIMED;
                            lc._rightBuffer = buffer;
                            lc._rightFoundAt = foundAt2;
                            pageAddr2 = buffer.getPageAddress();
                        }

                        if (lc._leftBuffer.isIndexPage()) {
                            Debug.$assert0.t(lc._rightBuffer.isIndexPage() && depth > 0);
                            //
                            // Come down to the left of the key.
                            //
                            int p1 = lc._leftBuffer.previousKeyBlock(foundAt1);
                            int p2 = lc._rightBuffer.previousKeyBlock(foundAt2);

                            Debug.$assert0.t(p1 != -1 && p2 != -1);
                            pageAddr1 = lc._leftBuffer.getPointer(p1);
                            pageAddr2 = lc._rightBuffer.getPointer(p2);

                        } else {
                            Debug.$assert0.t(depth == 0);
                            break;
                        }
                    }

                    LevelCache lc = _levelCache[0];
                    if (removeOnlyAntiValue
                            & !isKeyRangeAntiValue(lc._leftBuffer, lc._leftFoundAt, lc._rightBuffer, lc._rightFoundAt)) {
                        result = false;
                        break;
                    }

                    if (fetchFirst) {
                        removeFetchFirst(lc._leftBuffer, lc._leftFoundAt, lc._rightBuffer, lc._rightFoundAt);
                    }
                    //
                    // We have fully delineated the subtree that
                    // needs to be removed. Now walk down the tree,
                    // stitching together the pages where necessary.
                    //
                    final long timestamp = timestamp();
                    for (int level = _cacheDepth; --level >= 0;) {
                        lc = _levelCache[level];
                        Buffer buffer1 = lc._leftBuffer;
                        Buffer buffer2 = lc._rightBuffer;
                        int foundAt1 = lc._leftFoundAt;
                        int foundAt2 = lc._rightFoundAt;
                        foundAt1 &= P_MASK;
                        foundAt2 &= P_MASK;

                        boolean needsReindex = false;
                        buffer1.writePageOnCheckpoint(timestamp);
                        if (buffer1 != buffer2) {
                            buffer2.writePageOnCheckpoint(timestamp);
                            //
                            // Deletion spans multiple pages at this level.
                            // We will need to join or rebalance the pages.
                            //
                            long leftGarbagePage = buffer1.getRightSibling();
                            _key.copyTo(_spareKey1);

                            // Before we remove the records in this range, we
                            // need to recover any LONG_RECORD pointers that
                            // are associated with keys in this range.
                            _volume.getStructure().harvestLongRecords(buffer1, foundAt1, Integer.MAX_VALUE);
                            _volume.getStructure().harvestLongRecords(buffer2, 0, foundAt2);

                            Debug.$assert0.t(_tree.isMine() && buffer1.isMine() && buffer2.isMine());
                            boolean rebalanced = buffer1.join(buffer2, foundAt1, foundAt2, _spareKey1, _spareKey2,
                                    _joinPolicy);
                            if (buffer1.isDataPage()) {
                                _tree.bumpChangeCount();
                            }

                            buffer1.setDirtyAtTimestamp(timestamp);
                            buffer2.setDirtyAtTimestamp(timestamp);

                            long rightGarbagePage = buffer1.getRightSibling();

                            if (rightGarbagePage != leftGarbagePage) {
                                // here we just remember the page boundaries
                                // that will need to be deallocated.
                                lc._deallocLeftPage = leftGarbagePage;
                                lc._deallocRightPage = rightGarbagePage;
                                deallocationRequired = true;
                            }

                            if (rebalanced) {
                                //
                                // If the join operation was not able to
                                // coalesce the two pages into one, then we need
                                // to re-index the new first key of the second
                                // page.
                                //
                                // We have either a quick way to do this or a
                                // more complex way. If there is a single parent
                                // page in the index for the two re-balanced
                                // pages, and if the key to be reinserted fits
                                // in that parent page, then all we need to do
                                // is insert it. Otherwise, we will need to
                                // split the page above us, and that will
                                // potentially result in additional buffer
                                // reservations. Because that could force a
                                // retry at a bad time, in that case we defer
                                // the re-insertion of the index key until
                                // after all the current claims are released.
                                //
                                needsReindex = true;
                                if (level < _cacheDepth - 1) {
                                    LevelCache parentLc = _levelCache[level + 1];
                                    Buffer buffer = parentLc._leftBuffer;

                                    Debug.$assert0.t(buffer != null);
                                    if (parentLc._rightBuffer == buffer) {
                                        int foundAt = buffer.findKey(_spareKey1);
                                        Debug.$assert0.t((foundAt & EXACT_MASK) == 0);
                                        // Try it the simple way
                                        _value.setPointerValue(buffer2.getPageAddress());
                                        _value.setPointerPageType(buffer2.getPageType());
                                        _rawValueWriter.init(_value);
                                        int fit = buffer.putValue(_spareKey1, _rawValueWriter, foundAt, false);

                                        // If it worked then we're done.
                                        if (fit != -1) {
                                            needsReindex = false;
                                            buffer.setDirtyAtTimestamp(timestamp);
                                        }
                                    }
                                }
                                if (needsReindex) {
                                    _spareKey1.copyTo(_spareKey2);
                                    _value.setPointerValue(buffer2.getPageAddress());
                                    _value.setPointerPageType(buffer2.getPageType());

                                    storeInternal(_spareKey2, _value, level + 1, StoreOptions.NONE);
                                    needsReindex = false;
                                }
                            }

                            result = true;
                        } else if (foundAt1 != foundAt2) {
                            Debug.$assert0.t(foundAt2 > foundAt1);
                            _key.copyTo(_spareKey1);
                            //
                            // Before we remove these records, we need to
                            // recover any LONG_RECORD pointers that may be
                            // associated with keys in this range.
                            //
                            _volume.getStructure().harvestLongRecords(buffer1, foundAt1, foundAt2);
                            result |= buffer1.removeKeys(foundAt1, foundAt2, _spareKey1);
                            if (buffer1.isDataPage() && result) {
                                _tree.bumpChangeCount();
                            }
                            buffer1.setDirtyAtTimestamp(timestamp);
                        }

                        if (level < _cacheDepth - 1) {
                            removeKeyRangeReleaseLevel(level + 1);
                        }
                    }
                    break;
                } catch (RetryException re) {
                    // handled below by releasing claims and retrying
                } finally {
                    //
                    // Release all buffers.
                    //
                    for (int level = _cacheDepth; --level >= depth;) {
                        removeKeyRangeReleaseLevel(level);
                    }

                    if (treeClaimAcquired) {
                        if (treeWriterClaimRequired) {
                            _tree.bumpGeneration();
                        }
                        _treeHolder.release();
                        treeClaimAcquired = false;
                    }
                }
                /*
                 * Having released all prior claims, now acquire an exclusive
                 * claim on the Tree.
                 */
                if (treeWriterClaimRequired) {
                    if (!_treeHolder.claim(true)) {
                        Debug.$assert0.t(false);
                        throw new InUseException("Thread " + Thread.currentThread().getName()
                                + " failed to get reader claim on " + _tree);
                    }
                    treeClaimAcquired = true;
                }
            }

            while (deallocationRequired) {
                long left = -1;
                long right = -1;
                for (int level = _cacheDepth; --level >= 0;) {
                    LevelCache lc = _levelCache[level];
                    left = lc._deallocLeftPage;
                    right = lc._deallocRightPage;
                    if (left != 0) {
                        _volume.getStructure().deallocateGarbageChain(left, right);
                        lc._deallocLeftPage = 0;
                        lc._deallocRightPage = 0;
                    }
                }
                // If we successfully finish the loop then we're done
                deallocationRequired = false;
                break;
            }
        } finally {
            if (treeClaimAcquired) {
                if (treeWriterClaimRequired) {
                    _tree.bumpGeneration();
                }
                _treeHolder.release();
                treeClaimAcquired = false;
            }
        }

        _volume.getStatistics().bumpRemoveCounter();
        _tree.getStatistics().bumpRemoveCounter();
        if (fetchFirst) {
            _volume.getStatistics().bumpFetchCounter();
            _tree.getStatistics().bumpFetchCounter();
        }
        return result;
    }