/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.hyracks.storage.am.lsm.btree.impls;

import java.io.IOException;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.common.data.accessors.ITupleReference;
import org.apache.hyracks.storage.am.bloomfilter.impls.BloomCalculations;
import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilter;
import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilterFactory;
import org.apache.hyracks.storage.am.bloomfilter.impls.BloomFilterSpecification;
import org.apache.hyracks.storage.am.btree.impls.BTree;
import org.apache.hyracks.storage.am.btree.impls.BTree.BTreeBulkLoader;
import org.apache.hyracks.storage.am.btree.impls.RangePredicate;
import org.apache.hyracks.storage.am.common.api.IFreePageManager;
import org.apache.hyracks.storage.am.common.api.IIndexBulkLoader;
import org.apache.hyracks.storage.am.common.api.IIndexCursor;
import org.apache.hyracks.storage.am.common.api.IIndexOperationContext;
import org.apache.hyracks.storage.am.common.api.IModificationOperationCallback;
import org.apache.hyracks.storage.am.common.api.ISearchOperationCallback;
import org.apache.hyracks.storage.am.common.api.ISearchPredicate;
import org.apache.hyracks.storage.am.common.api.ITreeIndex;
import org.apache.hyracks.storage.am.common.api.ITreeIndexCursor;
import org.apache.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
import org.apache.hyracks.storage.am.common.api.ITwoPCIndexBulkLoader;
import org.apache.hyracks.storage.am.common.api.IndexException;
import org.apache.hyracks.storage.am.common.api.TreeIndexException;
import org.apache.hyracks.storage.am.common.impls.NoOpOperationCallback;
import org.apache.hyracks.storage.am.common.ophelpers.IndexOperation;
import org.apache.hyracks.storage.am.common.ophelpers.MultiComparator;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponent;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMComponentFactory;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMHarness;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperation;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationCallback;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIOOperationScheduler;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexAccessorInternal;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexFileManager;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndexOperationContext;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicy;
import org.apache.hyracks.storage.am.lsm.common.api.ILSMOperationTracker;
import org.apache.hyracks.storage.am.lsm.common.api.ITwoPCIndex;
import org.apache.hyracks.storage.am.lsm.common.api.LSMOperationType;
import org.apache.hyracks.storage.am.lsm.common.impls.AbstractLSMIndex;
import org.apache.hyracks.storage.am.lsm.common.impls.BlockingIOOperationCallbackWrapper;
import org.apache.hyracks.storage.am.lsm.common.impls.ExternalIndexHarness;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMComponentFileReferences;
import org.apache.hyracks.storage.am.lsm.common.impls.LSMTreeIndexAccessor;
import org.apache.hyracks.storage.am.lsm.common.impls.TreeIndexFactory;
import org.apache.hyracks.storage.common.buffercache.IBufferCache;
import org.apache.hyracks.storage.common.file.IFileMapProvider;

public class ExternalBTreeWithBuddy extends AbstractLSMIndex implements ITreeIndex, ITwoPCIndex {

    // For creating merge disk components
    private final LSMBTreeWithBuddyDiskComponentFactory componentFactory;

    private final LSMBTreeWithBuddyDiskComponentFactory bulkComponentFactory;

    private final IBinaryComparatorFactory[] btreeCmpFactories;
    private final IBinaryComparatorFactory[] buddyBtreeCmpFactories;
    private final int[] buddyBTreeFields;

    // Common for in-memory and on-disk components.
    private final ITreeIndexFrameFactory btreeInteriorFrameFactory;
    private final ITreeIndexFrameFactory btreeLeafFrameFactory;
    private final ITreeIndexFrameFactory buddyBtreeLeafFrameFactory;

    // A second disk component list that will be used when a transaction is
    // committed and will be seen by subsequent accessors
    private final List<ILSMComponent> secondDiskComponents;
    private int version = -1;

    public ExternalBTreeWithBuddy(ITreeIndexFrameFactory btreeInteriorFrameFactory,
            ITreeIndexFrameFactory btreeLeafFrameFactory, ITreeIndexFrameFactory buddyBtreeLeafFrameFactory,
            IBufferCache diskBufferCache, ILSMIndexFileManager fileManager,
            TreeIndexFactory<BTree> bulkLoadBTreeFactory, TreeIndexFactory<BTree> copyBtreeFactory,
            TreeIndexFactory<BTree> buddyBtreeFactory, BloomFilterFactory bloomFilterFactory,
            IFileMapProvider diskFileMapProvider, double bloomFilterFalsePositiveRate, ILSMMergePolicy mergePolicy,
            ILSMOperationTracker opTracker, ILSMIOOperationScheduler ioScheduler, ILSMIOOperationCallback ioOpCallback,
            IBinaryComparatorFactory[] btreeCmpFactories, IBinaryComparatorFactory[] buddyBtreeCmpFactories,
            int[] buddyBTreeFields, int version, boolean durable) {
        super(diskBufferCache, fileManager, diskFileMapProvider, bloomFilterFalsePositiveRate, mergePolicy, opTracker,
                ioScheduler, ioOpCallback, durable);
        this.btreeCmpFactories = btreeCmpFactories;
        this.buddyBtreeCmpFactories = buddyBtreeCmpFactories;
        this.buddyBTreeFields = buddyBTreeFields;
        this.btreeInteriorFrameFactory = btreeInteriorFrameFactory;
        this.btreeLeafFrameFactory = btreeLeafFrameFactory;
        this.buddyBtreeLeafFrameFactory = buddyBtreeLeafFrameFactory;
        this.componentFactory = new LSMBTreeWithBuddyDiskComponentFactory(copyBtreeFactory, buddyBtreeFactory,
                bloomFilterFactory);
        this.bulkComponentFactory = new LSMBTreeWithBuddyDiskComponentFactory(bulkLoadBTreeFactory, buddyBtreeFactory,
                bloomFilterFactory);
        this.secondDiskComponents = new LinkedList<ILSMComponent>();
        this.version = version;
    }

    @Override
    public void create() throws HyracksDataException {
        if (isActivated) {
            throw new HyracksDataException("Failed to create the index since it is activated.");
        }
        fileManager.deleteDirs();
        fileManager.createDirs();
        diskComponents.clear();
        secondDiskComponents.clear();
    }

    @Override
    public void activate() throws HyracksDataException {
        if (isActivated) {
            throw new HyracksDataException("Failed to activate the index since it is already activated.");
        }

        if (diskComponents.size() == 0 && secondDiskComponents.size() == 0) {
            //First time activation
            List<LSMComponentFileReferences> validFileReferences;
            try {
                validFileReferences = fileManager.cleanupAndGetValidFiles();
            } catch (IndexException e) {
                throw new HyracksDataException(e);
            }
            for (LSMComponentFileReferences lsmComonentFileReference : validFileReferences) {
                LSMBTreeWithBuddyDiskComponent component;
                try {
                    component = createDiskComponent(componentFactory,
                            lsmComonentFileReference.getInsertIndexFileReference(),
                            lsmComonentFileReference.getDeleteIndexFileReference(),
                            lsmComonentFileReference.getBloomFilterFileReference(), false);
                } catch (IndexException e) {
                    throw new HyracksDataException(e);
                }
                diskComponents.add(component);
                secondDiskComponents.add(component);
            }
            ((ExternalIndexHarness) lsmHarness).indexFirstTimeActivated();
        } else {
            // This index has been opened before or is brand new with no
            // components. It should also maintain the version pointer
            for (ILSMComponent c : diskComponents) {
                LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
                BTree btree = component.getBTree();
                BTree buddyBtree = component.getBuddyBTree();
                BloomFilter bloomFilter = component.getBloomFilter();
                btree.activate();
                buddyBtree.activate();
                bloomFilter.activate();
            }
            for (ILSMComponent c : secondDiskComponents) {
                // Only activate non shared components
                if (!diskComponents.contains(c)) {
                    LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
                    BTree btree = component.getBTree();
                    BTree buddyBtree = component.getBuddyBTree();
                    BloomFilter bloomFilter = component.getBloomFilter();
                    btree.activate();
                    buddyBtree.activate();
                    bloomFilter.activate();
                }
            }
        }
        isActivated = true;
    }

    @Override
    public void clear() throws HyracksDataException {
        if (!isActivated) {
            throw new HyracksDataException("Failed to clear the index since it is not activated.");
        }
        ((ExternalIndexHarness) lsmHarness).indexClear();

        for (ILSMComponent c : diskComponents) {
            LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
            component.getBTree().deactivate();
            component.getBuddyBTree().deactivate();
            component.getBloomFilter().deactivate();
            component.getBTree().destroy();
            component.getBloomFilter().destroy();
            component.getBuddyBTree().destroy();
            // Remove from second list to avoid destroying twice
            secondDiskComponents.remove(c);
        }

        for (ILSMComponent c : secondDiskComponents) {
            LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
            component.getBTree().deactivate();
            component.getBloomFilter().deactivate();
            component.getBuddyBTree().deactivate();
            component.getBTree().destroy();
            component.getBloomFilter().destroy();
            component.getBuddyBTree().destroy();
        }

        diskComponents.clear();
        secondDiskComponents.clear();
        version = -1;
    }

    @Override
    public void deactivate() throws HyracksDataException {
        deactivate(true);
    }

    @Override
    public void destroy() throws HyracksDataException {
        if (isActivated) {
            throw new HyracksDataException("Failed to destroy the index since it is activated.");
        }
        for (ILSMComponent c : diskComponents) {
            LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
            component.getBTree().destroy();
            component.getBuddyBTree().destroy();
            component.getBloomFilter().destroy();
            // Remove from second list to avoid destroying twice
            secondDiskComponents.remove(c);
        }
        for (ILSMComponent c : secondDiskComponents) {
            LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
            component.getBTree().destroy();
            component.getBuddyBTree().destroy();
            component.getBloomFilter().destroy();
        }
        diskComponents.clear();
        secondDiskComponents.clear();
        fileManager.deleteDirs();
        version = -1;
    }

    @Override
    public ILSMIndexAccessorInternal createAccessor(IModificationOperationCallback modificationCallback,
            ISearchOperationCallback searchCallback) throws HyracksDataException {
        return new LSMBTreeWithBuddyAccessor(lsmHarness, createOpContext(searchCallback, version));
    }

    @Override
    public void validate() throws HyracksDataException {
        throw new UnsupportedOperationException("Validation not implemented for LSM B-Trees with Buddy B-Tree.");
    }

    @Override
    public long getMemoryAllocationSize() {
        return 0;
    }

    // The subsume merged components is overridden to account for:
    // Maintaining two versions of the index
    @Override
    public void subsumeMergedComponents(ILSMComponent newComponent, List<ILSMComponent> mergedComponents)
            throws HyracksDataException {
        List<ILSMComponent> newerList;
        List<ILSMComponent> olderList;
        if (version == 0) {
            newerList = diskComponents;
            olderList = secondDiskComponents;
        } else {
            newerList = secondDiskComponents;
            olderList = diskComponents;
        }

        // Check if merge will affect the older list
        if (olderList.containsAll(mergedComponents)) {
            int swapIndex = olderList.indexOf(mergedComponents.get(0));
            olderList.removeAll(mergedComponents);
            olderList.add(swapIndex, newComponent);
        }

        // The new list will always have all the merged components
        int swapIndex = newerList.indexOf(mergedComponents.get(0));
        newerList.removeAll(mergedComponents);
        newerList.add(swapIndex, newComponent);
    }

    // For initial load
    @Override
    public IIndexBulkLoader createBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
            boolean checkIfEmptyIndex) throws TreeIndexException {
        try {
            return new LSMTwoPCBTreeWithBuddyBulkLoader(fillLevel, verifyInput, 0, checkIfEmptyIndex, false);
        } catch (HyracksDataException e) {
            throw new TreeIndexException(e);
        }
    }

    // For transaction bulk load <- could consolidate with the above method ->
    @Override
    public IIndexBulkLoader createTransactionBulkLoader(float fillLevel, boolean verifyInput, long numElementsHint,
            boolean checkIfEmptyIndex) throws TreeIndexException {
        try {
            return new LSMTwoPCBTreeWithBuddyBulkLoader(fillLevel, verifyInput, numElementsHint, checkIfEmptyIndex,
                    true);
        } catch (HyracksDataException e) {
            throw new TreeIndexException(e);
        }
    }

    @Override
    public void modify(IIndexOperationContext ictx, ITupleReference tuple) throws HyracksDataException, IndexException {
        throw new UnsupportedOperationException("tuple modify not supported in LSM-Disk-Only-BTree");
    }

    @Override
    public void search(ILSMIndexOperationContext ictx, IIndexCursor cursor, ISearchPredicate pred)
            throws HyracksDataException, IndexException {
        ExternalBTreeWithBuddyOpContext ctx = (ExternalBTreeWithBuddyOpContext) ictx;
        List<ILSMComponent> operationalComponents = ictx.getComponentHolder();

        LSMBTreeWithBuddyCursorInitialState initialState = new LSMBTreeWithBuddyCursorInitialState(
                btreeInteriorFrameFactory, btreeLeafFrameFactory, buddyBtreeLeafFrameFactory, lsmHarness,
                MultiComparator.create(btreeCmpFactories), MultiComparator.create(buddyBtreeCmpFactories),
                ctx.searchCallback, operationalComponents);

        cursor.open(initialState, pred);
    }

    @Override
    public void scheduleFlush(ILSMIndexOperationContext ctx, ILSMIOOperationCallback callback)
            throws HyracksDataException {
        throw new UnsupportedOperationException("flush not supported in LSM-Disk-Only-BTree");
    }

    @Override
    public ILSMComponent flush(ILSMIOOperation operation) throws HyracksDataException, IndexException {
        throw new UnsupportedOperationException("flush not supported in LSM-Disk-Only-BTree");
    }

    protected LSMComponentFileReferences getMergeTargetFileName(List<ILSMComponent> mergingDiskComponents)
            throws HyracksDataException {
        BTree firstTree = ((LSMBTreeWithBuddyDiskComponent) mergingDiskComponents.get(0)).getBTree();
        BTree lastTree = ((LSMBTreeWithBuddyDiskComponent) mergingDiskComponents.get(mergingDiskComponents.size() - 1))
                .getBTree();
        FileReference firstFile = diskFileMapProvider.lookupFileName(firstTree.getFileId());
        FileReference lastFile = diskFileMapProvider.lookupFileName(lastTree.getFileId());
        LSMComponentFileReferences fileRefs = fileManager.getRelMergeFileReference(firstFile.getFile().getName(),
                lastFile.getFile().getName());
        return fileRefs;
    }

    @Override
    public void scheduleMerge(ILSMIndexOperationContext ctx, ILSMIOOperationCallback callback)
            throws HyracksDataException, IndexException {
        ILSMIndexOperationContext bctx = createOpContext(NoOpOperationCallback.INSTANCE, 0);
        bctx.setOperation(IndexOperation.MERGE);
        List<ILSMComponent> mergingComponents = ctx.getComponentHolder();
        ITreeIndexCursor cursor = new LSMBTreeWithBuddySortedCursor(bctx, buddyBTreeFields);
        LSMComponentFileReferences relMergeFileRefs = getMergeTargetFileName(mergingComponents);
        ILSMIndexAccessorInternal accessor = new LSMBTreeWithBuddyAccessor(lsmHarness, bctx);

        // Since we have two lists of components, to tell whether we need to
        // keep deleted tuples, we need to know
        // which list to check against and we need to synchronize for this
        boolean keepDeleteTuples = false;
        if (version == 0) {
            keepDeleteTuples = mergingComponents.get(mergingComponents.size() - 1) != diskComponents.get(diskComponents
                    .size() - 1);
        } else {
            keepDeleteTuples = mergingComponents.get(mergingComponents.size() - 1) != secondDiskComponents
                    .get(secondDiskComponents.size() - 1);
        }

        ioScheduler.scheduleOperation(new LSMBTreeWithBuddyMergeOperation(accessor, mergingComponents, cursor,
                relMergeFileRefs.getInsertIndexFileReference(), relMergeFileRefs.getDeleteIndexFileReference(),
                relMergeFileRefs.getBloomFilterFileReference(), callback, fileManager.getBaseDir(), keepDeleteTuples));
    }

    // This method creates the appropriate opContext for the targeted version
    public ExternalBTreeWithBuddyOpContext createOpContext(ISearchOperationCallback searchCallback, int targetVersion) {
        return new ExternalBTreeWithBuddyOpContext(btreeCmpFactories, buddyBtreeCmpFactories, searchCallback,
                targetVersion);
    }

    @Override
    public ILSMComponent merge(ILSMIOOperation operation) throws HyracksDataException, IndexException {
        LSMBTreeWithBuddyMergeOperation mergeOp = (LSMBTreeWithBuddyMergeOperation) operation;
        ITreeIndexCursor cursor = mergeOp.getCursor();
        ISearchPredicate btreeSearchPred = new RangePredicate(null, null, true, true, null, null);
        ILSMIndexOperationContext opCtx = ((LSMBTreeWithBuddySortedCursor) cursor).getOpCtx();
        opCtx.getComponentHolder().addAll(mergeOp.getMergingComponents());
        search(opCtx, cursor, btreeSearchPred);

        LSMBTreeWithBuddyDiskComponent mergedComponent = createDiskComponent(componentFactory,
                mergeOp.getBTreeMergeTarget(), mergeOp.getBuddyBTreeMergeTarget(), mergeOp.getBloomFilterMergeTarget(),
                true);

        // In case we must keep the deleted-keys BuddyBTrees, then they must be
        // merged *before* merging the b-trees so that
        // lsmHarness.endSearch() is called once when the b-trees have been
        // merged.

        if (mergeOp.isKeepDeletedTuples()) {
            // Keep the deleted tuples since the oldest disk component is not
            // included in the merge operation
            LSMBuddyBTreeMergeCursor buddyBtreeCursor = new LSMBuddyBTreeMergeCursor(opCtx);
            search(opCtx, buddyBtreeCursor, btreeSearchPred);

            BTree buddyBtree = mergedComponent.getBuddyBTree();
            IIndexBulkLoader buddyBtreeBulkLoader = buddyBtree.createBulkLoader(1.0f, true, 0L, false);

            long numElements = 0L;
            for (int i = 0; i < mergeOp.getMergingComponents().size(); ++i) {
                numElements += ((LSMBTreeWithBuddyDiskComponent) mergeOp.getMergingComponents().get(i))
                        .getBloomFilter().getNumElements();
            }

            int maxBucketsPerElement = BloomCalculations.maxBucketsPerElement(numElements);
            BloomFilterSpecification bloomFilterSpec = BloomCalculations.computeBloomSpec(maxBucketsPerElement,
                    bloomFilterFalsePositiveRate);
            IIndexBulkLoader builder = mergedComponent.getBloomFilter().createBuilder(numElements,
                    bloomFilterSpec.getNumHashes(), bloomFilterSpec.getNumBucketsPerElements());

            try {
                while (buddyBtreeCursor.hasNext()) {
                    buddyBtreeCursor.next();
                    ITupleReference tuple = buddyBtreeCursor.getTuple();
                    buddyBtreeBulkLoader.add(tuple);
                    builder.add(tuple);
                }
            } finally {
                buddyBtreeCursor.close();
                builder.end();
            }
            buddyBtreeBulkLoader.end();
        }

        IIndexBulkLoader bulkLoader = mergedComponent.getBTree().createBulkLoader(1.0f, false, 0L, false);
        try {
            while (cursor.hasNext()) {
                cursor.next();
                ITupleReference frameTuple = cursor.getTuple();
                bulkLoader.add(frameTuple);
            }
        } finally {
            cursor.close();
        }
        bulkLoader.end();
        return mergedComponent;
    }

    @Override
    public void getOperationalComponents(ILSMIndexOperationContext ctx) {
        List<ILSMComponent> operationalComponents = ctx.getComponentHolder();
        List<ILSMComponent> immutableComponents;
        // Identify current list in case of a merge
        if (version == 0) {
            immutableComponents = diskComponents;
        } else {
            immutableComponents = secondDiskComponents;
        }

        ExternalBTreeWithBuddyOpContext opCtx = (ExternalBTreeWithBuddyOpContext) ctx;
        operationalComponents.clear();
        switch (ctx.getOperation()) {
            case SEARCH:
                if (opCtx.getTargetIndexVersion() == 0) {
                    operationalComponents.addAll(diskComponents);
                } else {
                    operationalComponents.addAll(secondDiskComponents);
                }
                break;
            case MERGE:
                operationalComponents.addAll(ctx.getComponentsToBeMerged());
                break;
            case FULL_MERGE:
                operationalComponents.addAll(immutableComponents);
            case REPLICATE:
                operationalComponents.addAll(ctx.getComponentsToBeReplicated());
                break;
            case FLUSH:
                // Do nothing. this is left here even though the index never
                // performs flushes because a flush is triggered by
                // dataset lifecycle manager when closing an index. Having no
                // components is a no operation
                break;
            default:
                throw new UnsupportedOperationException("Operation " + ctx.getOperation() + " not supported.");
        }
    }

    @Override
    public void markAsValid(ILSMComponent lsmComponent) throws HyracksDataException {
        LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) lsmComponent;
        // Flush the bloom filter first.
        int fileId = component.getBloomFilter().getFileId();
        IBufferCache bufferCache = component.getBTree().getBufferCache();
        int startPage = 0;
        int maxPage = component.getBloomFilter().getNumPages();
        forceFlushDirtyPages(bufferCache, fileId, startPage, maxPage);
        forceFlushDirtyPages(component.getBTree());
        markAsValidInternal(component.getBTree());
        forceFlushDirtyPages(component.getBuddyBTree());
        markAsValidInternal(component.getBuddyBTree());
    }

    // This function is used when a new component is to be committed -- is
    // called by the harness.
    @Override
    public void commitTransactionDiskComponent(ILSMComponent newComponent) throws HyracksDataException {

        // determine which list is the new one and flip the pointer
        List<ILSMComponent> newerList;
        List<ILSMComponent> olderList;
        if (version == 0) {
            newerList = diskComponents;
            olderList = secondDiskComponents;
            version = 1;
        } else {
            newerList = secondDiskComponents;
            olderList = diskComponents;
            version = 0;
        }
        // Remove components from list
        olderList.clear();
        // Add components
        olderList.addAll(newerList);
        if (newComponent != null) {
            // Add it to the list
            olderList.add(0, newComponent);
        }
    }

    @Override
    public void deactivate(boolean flushOnExit) throws HyracksDataException {
        if (!isActivated) {
            throw new HyracksDataException("Failed to deactivate the index since it is already deactivated.");
        }

        if (flushOnExit) {
            BlockingIOOperationCallbackWrapper cb = new BlockingIOOperationCallbackWrapper(ioOpCallback);
            cb.afterFinalize(LSMOperationType.FLUSH, null);
        }
        // Even though, we deactivate the index, we don't exit components or
        // modify any of the lists to make sure they
        // are there if the index was opened again

        for (ILSMComponent c : diskComponents) {
            LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
            BTree btree = component.getBTree();
            BTree buddyBtree = component.getBuddyBTree();
            BloomFilter bloomFilter = component.getBloomFilter();
            btree.deactivate();
            buddyBtree.deactivate();
            bloomFilter.deactivate();
        }
        for (ILSMComponent c : secondDiskComponents) {
            // Only deactivate non shared components
            if (!diskComponents.contains(c)) {
                LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) c;
                BTree btree = component.getBTree();
                BTree buddyBtree = component.getBuddyBTree();
                BloomFilter bloomFilter = component.getBloomFilter();
                btree.deactivate();
                buddyBtree.deactivate();
                bloomFilter.deactivate();
            }
        }
        isActivated = false;
    }

    @Override
    public ITreeIndexFrameFactory getLeafFrameFactory() {
        return btreeLeafFrameFactory;
    }

    @Override
    public ITreeIndexFrameFactory getInteriorFrameFactory() {
        return btreeInteriorFrameFactory;
    }

    @Override
    public IFreePageManager getFreePageManager() {
        // This method should never be called for disk only indexes
        return null;
    }

    @Override
    public int getFieldCount() {
        return btreeCmpFactories.length;
    }

    @Override
    public int getRootPageId() {
        // This method should never be called for this index
        return 0;
    }

    @Override
    public int getFileId() {
        // This method should never be called for this index
        return 0;
    }

    @Override
    public IBinaryComparatorFactory[] getComparatorFactories() {
        return btreeCmpFactories;
    }

    private LSMBTreeWithBuddyDiskComponent createDiskComponent(ILSMComponentFactory factory,
            FileReference insertFileRef, FileReference deleteFileRef, FileReference bloomFilterFileRef,
            boolean createComponent) throws HyracksDataException, IndexException {
        // Create new instance.
        LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) factory
                .createLSMComponentInstance(new LSMComponentFileReferences(insertFileRef, deleteFileRef,
                        bloomFilterFileRef));
        if (createComponent) {
            component.getBTree().create();
            component.getBuddyBTree().create();
            component.getBloomFilter().create();
        }

        component.getBTree().activate();
        component.getBuddyBTree().activate();
        component.getBloomFilter().activate();
        return component;
    }

    // even though the index doesn't support record level modification, the
    // accessor will try to do it
    // we could throw the exception here but we don't. it will eventually be
    // thrown by the index itself
    public class LSMBTreeWithBuddyAccessor extends LSMTreeIndexAccessor {
        public LSMBTreeWithBuddyAccessor(ILSMHarness lsmHarness, ILSMIndexOperationContext ctx) {
            super(lsmHarness, ctx);
        }

        @Override
        public ITreeIndexCursor createSearchCursor(boolean exclusive) {
            return new LSMBTreeWithBuddySearchCursor(ctx, buddyBTreeFields);
        }

        public MultiComparator getBTreeMultiComparator() {
            ExternalBTreeWithBuddyOpContext concreteCtx = (ExternalBTreeWithBuddyOpContext) ctx;
            return concreteCtx.getBTreeMultiComparator();
        }

        public MultiComparator getBodyBTreeMultiComparator() {
            ExternalBTreeWithBuddyOpContext concreteCtx = (ExternalBTreeWithBuddyOpContext) ctx;
            return concreteCtx.getBuddyBTreeMultiComparator();
        }
    }

    // The bulk loader used for both initial loading and transaction
    // modifications
    public class LSMTwoPCBTreeWithBuddyBulkLoader implements IIndexBulkLoader, ITwoPCIndexBulkLoader {
        private final ILSMComponent component;
        private final BTreeBulkLoader btreeBulkLoader;
        private final BTreeBulkLoader buddyBtreeBulkLoader;
        private final IIndexBulkLoader builder;
        private boolean cleanedUpArtifacts = false;
        private boolean isEmptyComponent = true;
        private boolean endedBloomFilterLoad = false;
        private final boolean isTransaction;

        public LSMTwoPCBTreeWithBuddyBulkLoader(float fillFactor, boolean verifyInput, long numElementsHint,
                boolean checkIfEmptyIndex, boolean isTransaction) throws TreeIndexException, HyracksDataException {
            this.isTransaction = isTransaction;
            // Create the appropriate target
            if (isTransaction) {
                try {
                    component = createTransactionTarget();
                } catch (HyracksDataException | IndexException e) {
                    throw new TreeIndexException(e);
                }
            } else {
                if (checkIfEmptyIndex && !isEmptyIndex()) {
                    throw new TreeIndexException("Cannot load an index that is not empty");
                }
                try {
                    component = createBulkLoadTarget();
                } catch (HyracksDataException | IndexException e) {
                    throw new TreeIndexException(e);
                }
            }

            // Create the three loaders
            btreeBulkLoader = (BTreeBulkLoader) ((LSMBTreeWithBuddyDiskComponent) component).getBTree()
                    .createBulkLoader(fillFactor, verifyInput, numElementsHint, false);
            buddyBtreeBulkLoader = (BTreeBulkLoader) ((LSMBTreeWithBuddyDiskComponent) component).getBuddyBTree()
                    .createBulkLoader(fillFactor, verifyInput, numElementsHint, false);
            int maxBucketsPerElement = BloomCalculations.maxBucketsPerElement(numElementsHint);
            BloomFilterSpecification bloomFilterSpec = BloomCalculations.computeBloomSpec(maxBucketsPerElement,
                    bloomFilterFalsePositiveRate);
            builder = ((LSMBTreeWithBuddyDiskComponent) component).getBloomFilter().createBuilder(numElementsHint,
                    bloomFilterSpec.getNumHashes(), bloomFilterSpec.getNumBucketsPerElements());
        }

        @Override
        public void add(ITupleReference tuple) throws IndexException, HyracksDataException {
            try {
                btreeBulkLoader.add(tuple);
            } catch (IndexException | HyracksDataException | RuntimeException e) {
                cleanupArtifacts();
                throw e;
            }
            if (isEmptyComponent) {
                isEmptyComponent = false;
            }
        }

        // This is made public in case of a failure, it is better to delete all
        // created artifacts.
        public void cleanupArtifacts() throws HyracksDataException {
            if (!cleanedUpArtifacts) {
                cleanedUpArtifacts = true;
                try {
                    ((LSMBTreeWithBuddyDiskComponent) component).getBTree().deactivate();
                } catch (Exception e) {

                }
                ((LSMBTreeWithBuddyDiskComponent) component).getBTree().destroy();
                try {
                    ((LSMBTreeWithBuddyDiskComponent) component).getBuddyBTree().deactivate();
                } catch (Exception e) {

                }
                ((LSMBTreeWithBuddyDiskComponent) component).getBuddyBTree().destroy();
                try {
                    ((LSMBTreeWithBuddyDiskComponent) component).getBloomFilter().deactivate();
                } catch (Exception e) {

                }
                ((LSMBTreeWithBuddyDiskComponent) component).getBloomFilter().destroy();
            }
        }

        @Override
        public void end() throws HyracksDataException, IndexException {
            if (!cleanedUpArtifacts) {
                if (!endedBloomFilterLoad) {
                    builder.end();
                    endedBloomFilterLoad = true;
                }
                btreeBulkLoader.end();
                buddyBtreeBulkLoader.end();
                if (isEmptyComponent) {
                    cleanupArtifacts();
                } else if (isTransaction) {
                    // Since this is a transaction component, validate and
                    // deactivate. it could later be added or deleted
                    markAsValid(component);
                    BTree btree = ((LSMBTreeWithBuddyDiskComponent) component).getBTree();
                    BTree buddyBtree = ((LSMBTreeWithBuddyDiskComponent) component).getBuddyBTree();
                    BloomFilter bloomFilter = ((LSMBTreeWithBuddyDiskComponent) component).getBloomFilter();
                    btree.deactivate();
                    buddyBtree.deactivate();
                    bloomFilter.deactivate();
                } else {
                    lsmHarness.addBulkLoadedComponent(component);
                }
            }
        }

        @Override
        public void delete(ITupleReference tuple) throws IndexException, HyracksDataException {
            try {
                buddyBtreeBulkLoader.add(tuple);
                builder.add(tuple);
            } catch (IndexException | HyracksDataException | RuntimeException e) {
                cleanupArtifacts();
                throw e;
            }
            if (isEmptyComponent) {
                isEmptyComponent = false;
            }
        }

        @Override
        public void abort() {
            try {
                cleanupArtifacts();
            } catch (Exception e) {
            }
        }
    }

    protected ILSMComponent createBulkLoadTarget() throws HyracksDataException, IndexException {
        LSMComponentFileReferences componentFileRefs = fileManager.getRelFlushFileReference();
        return createDiskComponent(bulkComponentFactory, componentFileRefs.getInsertIndexFileReference(),
                componentFileRefs.getDeleteIndexFileReference(), componentFileRefs.getBloomFilterFileReference(), true);
    }

    // This method is used to create a target for a bulk modify operation. This
    // component must then eventually be either committed or deleted
    private ILSMComponent createTransactionTarget() throws HyracksDataException, IndexException {
        LSMComponentFileReferences componentFileRefs;
        try {
            componentFileRefs = fileManager.getNewTransactionFileReference();
        } catch (IOException e) {
            throw new HyracksDataException("Failed to create transaction components", e);
        }
        return createDiskComponent(bulkComponentFactory, componentFileRefs.getInsertIndexFileReference(),
                componentFileRefs.getDeleteIndexFileReference(), componentFileRefs.getBloomFilterFileReference(), true);
    }

    @Override
    public ILSMIndexAccessorInternal createAccessor(ISearchOperationCallback searchCallback, int targetIndexVersion)
            throws HyracksDataException {
        return new LSMBTreeWithBuddyAccessor(lsmHarness, createOpContext(searchCallback, targetIndexVersion));
    }

    // This function in an instance of this index is only used after a bulk load
    // is successful
    // it will therefore add the component to the first list and enter it.
    @Override
    public void addComponent(ILSMComponent c) throws HyracksDataException {
        if (version == 0) {
            diskComponents.add(0, c);
        } else if (version == 1) {
            secondDiskComponents.add(0, c);
        }
    }

    @Override
    public int getCurrentVersion() {
        return version;
    }

    @Override
    public List<ILSMComponent> getFirstComponentList() {
        return diskComponents;
    }

    @Override
    public List<ILSMComponent> getSecondComponentList() {
        return secondDiskComponents;
    }

    @Override
    public void commitTransaction() throws HyracksDataException, IndexException {
        LSMComponentFileReferences componentFileRefrences = fileManager.getTransactionFileReferenceForCommit();
        LSMBTreeWithBuddyDiskComponent component = null;
        if (componentFileRefrences != null) {
            component = createDiskComponent(componentFactory, componentFileRefrences.getInsertIndexFileReference(),
                    componentFileRefrences.getDeleteIndexFileReference(),
                    componentFileRefrences.getBloomFilterFileReference(), false);
        }
        ((ExternalIndexHarness) lsmHarness).addTransactionComponents(component);
    }

    @Override
    public void abortTransaction() throws TreeIndexException {
        try {
            fileManager.deleteTransactionFiles();
        } catch (HyracksDataException e) {
            throw new TreeIndexException(e);
        }
    }

    @Override
    public void recoverTransaction() throws TreeIndexException {
        try {
            fileManager.recoverTransaction();
        } catch (HyracksDataException e) {
            throw new TreeIndexException(e);
        }
    }

    @Override
    public boolean hasMemoryComponents() {
        return false;
    }

    @Override
    public boolean isPrimaryIndex() {
        return false;
    }
    
    @Override
    public Set<String> getLSMComponentPhysicalFiles(ILSMComponent lsmComponent) {
        Set<String> files = new HashSet<String>();

        LSMBTreeWithBuddyDiskComponent component = (LSMBTreeWithBuddyDiskComponent) lsmComponent;
        files.add(component.getBTree().getFileReference().toString());
        files.add(component.getBuddyBTree().getFileReference().toString());
        files.add(component.getBloomFilter().getFileReference().toString());

        return files;
    }
}