path: root/src/mongo/db/index/index_access_method.cpp

/**
*    Copyright (C) 2013-2014 MongoDB Inc.
*
*    This program is free software: you can redistribute it and/or  modify
*    it under the terms of the GNU Affero General Public License, version 3,
*    as published by the Free Software Foundation.
*
*    This program is distributed in the hope that it will be useful,
*    but WITHOUT ANY WARRANTY; without even the implied warranty of
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
*    GNU Affero General Public License for more details.
*
*    You should have received a copy of the GNU Affero General Public License
*    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*
*    As a special exception, the copyright holders give permission to link the
*    code of portions of this program with the OpenSSL library under certain
*    conditions as described in each individual source file and distribute
*    linked combinations including the program with the OpenSSL library. You
*    must comply with the GNU Affero General Public License in all respects for
*    all of the code used other than as permitted herein. If you modify file(s)
*    with this exception, you may extend this exception to your version of the
*    file(s), but you are not obligated to do so. If you do not wish to do so,
*    delete this exception statement from your version. If you delete this
*    exception statement from all source files in the program, then also delete
*    it in the license file.
*/

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kIndex

#include "mongo/platform/basic.h"

#include "mongo/db/index/btree_access_method.h"

#include <vector>

#include "mongo/base/error_codes.h"
#include "mongo/base/status.h"
#include "mongo/db/client.h"
#include "mongo/db/concurrency/write_conflict_exception.h"
#include "mongo/db/curop.h"
#include "mongo/db/jsobj.h"
#include "mongo/db/keypattern.h"
#include "mongo/db/operation_context.h"
#include "mongo/db/server_parameters.h"
#include "mongo/db/storage/storage_options.h"
#include "mongo/util/log.h"
#include "mongo/util/progress_meter.h"

namespace mongo {

using std::endl;
using std::set;
using std::vector;

MONGO_EXPORT_SERVER_PARAMETER(failIndexKeyTooLong, bool, true);

//
// Comparison for external sorter interface
//

// Defined in db/structure/btree/key.cpp
// XXX TODO: rename to something more descriptive, etc. etc.
int oldCompare(const BSONObj& l, const BSONObj& r, const Ordering& o);

class BtreeExternalSortComparison {
public:
    BtreeExternalSortComparison(const BSONObj& ordering, int version)
        : _ordering(Ordering::make(ordering)), _version(version) {
        invariant(version == 1 || version == 0);
    }

    typedef std::pair<BSONObj, RecordId> Data;

    int operator()(const Data& l, const Data& r) const {
        int x = (_version == 1 ? l.first.woCompare(r.first, _ordering, /*considerfieldname*/ false)
                               : oldCompare(l.first, r.first, _ordering));
        if (x) {
            return x;
        }
        return l.second.compare(r.second);
    }

private:
    const Ordering _ordering;
    const int _version;
};

IndexAccessMethod::IndexAccessMethod(IndexCatalogEntry* btreeState, SortedDataInterface* btree)
    : _btreeState(btreeState), _descriptor(btreeState->descriptor()), _newInterface(btree) {
    verify(0 == _descriptor->version() || 1 == _descriptor->version());
}

bool IndexAccessMethod::ignoreKeyTooLong(OperationContext* txn) {
    // Ignore this error if we're on a secondary or if the user requested it
    return !txn->isPrimaryFor(_btreeState->ns()) || !failIndexKeyTooLong;
}

// Find the keys for obj, put them in the tree pointing to loc
Status IndexAccessMethod::insert(OperationContext* txn,
                                 const BSONObj& obj,
                                 const RecordId& loc,
                                 const InsertDeleteOptions& options,
                                 int64_t* numInserted) {
    *numInserted = 0;

    BSONObjSet keys;
    // Delegate to the subclass.
    getKeys(obj, &keys);

    Status ret = Status::OK();
    for (BSONObjSet::const_iterator i = keys.begin(); i != keys.end(); ++i) {
        Status status = _newInterface->insert(txn, *i, loc, options.dupsAllowed);

        // Everything's OK, carry on.
        if (status.isOK()) {
            ++*numInserted;
            continue;
        }

        // Error cases.

        if (status.code() == ErrorCodes::KeyTooLong && ignoreKeyTooLong(txn)) {
            continue;
        }

        if (status.code() == ErrorCodes::DuplicateKeyValue) {
            // A document might be indexed multiple times during a background index build
            // if it moves ahead of the collection scan cursor (e.g. via an update).
            if (!_btreeState->isReady(txn)) {
                LOG(3) << "key " << *i << " already in index during background indexing (ok)";
                continue;
            }
        }

        // Clean up after ourselves.
        for (BSONObjSet::const_iterator j = keys.begin(); j != i; ++j) {
            removeOneKey(txn, *j, loc, options.dupsAllowed);
            *numInserted = 0;
        }

        return status;
    }

    if (*numInserted > 1) {
        _btreeState->setMultikey(txn);
    }

    return ret;
}

void IndexAccessMethod::removeOneKey(OperationContext* txn,
                                     const BSONObj& key,
                                     const RecordId& loc,
                                     bool dupsAllowed) {
    try {
        _newInterface->unindex(txn, key, loc, dupsAllowed);
    } catch (AssertionException& e) {
        log() << "Assertion failure: _unindex failed " << _descriptor->indexNamespace() << endl;
        log() << "Assertion failure: _unindex failed: " << e.what() << "  key:" << key.toString()
              << "  dl:" << loc;
        logContext();
    }
}

std::unique_ptr<SortedDataInterface::Cursor> IndexAccessMethod::newCursor(OperationContext* txn,
                                                                          bool isForward) const {
    return _newInterface->newCursor(txn, isForward);
}

std::unique_ptr<SortedDataInterface::Cursor> IndexAccessMethod::newRandomCursor(
    OperationContext* txn) const {
    return _newInterface->newRandomCursor(txn);
}

// Remove the provided doc from the index.
Status IndexAccessMethod::remove(OperationContext* txn,
                                 const BSONObj& obj,
                                 const RecordId& loc,
                                 const InsertDeleteOptions& options,
                                 int64_t* numDeleted) {
    BSONObjSet keys;
    getKeys(obj, &keys);
    *numDeleted = 0;

    for (BSONObjSet::const_iterator i = keys.begin(); i != keys.end(); ++i) {
        removeOneKey(txn, *i, loc, options.dupsAllowed);
        ++*numDeleted;
    }

    return Status::OK();
}

// Return keys in l that are not in r.
// Lifted basically verbatim from elsewhere.
static void setDifference(const BSONObjSet& l, const BSONObjSet& r, vector<BSONObj*>* diff) {
    // l and r must use the same ordering spec.
    verify(l.key_comp().order() == r.key_comp().order());
    BSONObjSet::const_iterator i = l.begin();
    BSONObjSet::const_iterator j = r.begin();
    while (1) {
        if (i == l.end())
            break;
        while (j != r.end() && j->woCompare(*i) < 0)
            j++;
        if (j == r.end() || i->woCompare(*j) != 0) {
            const BSONObj* jo = &*i;
            diff->push_back((BSONObj*)jo);
        }
        i++;
    }
}

Status IndexAccessMethod::initializeAsEmpty(OperationContext* txn) {
    return _newInterface->initAsEmpty(txn);
}

Status IndexAccessMethod::touch(OperationContext* txn, const BSONObj& obj) {
    BSONObjSet keys;
    getKeys(obj, &keys);

    std::unique_ptr<SortedDataInterface::Cursor> cursor(_newInterface->newCursor(txn));
    for (BSONObjSet::const_iterator i = keys.begin(); i != keys.end(); ++i) {
        cursor->seekExact(*i);
    }

    return Status::OK();
}


Status IndexAccessMethod::touch(OperationContext* txn) const {
    return _newInterface->touch(txn);
}

RecordId IndexAccessMethod::findSingle(OperationContext* txn, const BSONObj& key) const {
    std::unique_ptr<SortedDataInterface::Cursor> cursor(_newInterface->newCursor(txn));
    const auto requestedInfo = kDebugBuild ? SortedDataInterface::Cursor::kKeyAndLoc
                                           : SortedDataInterface::Cursor::kWantLoc;
    if (auto kv = cursor->seekExact(key, requestedInfo)) {
        // StorageEngine should guarantee these.
        dassert(!kv->loc.isNull());
        dassert(kv->key.woCompare(key, /*order*/ BSONObj(), /*considerFieldNames*/ false) == 0);

        return kv->loc;
    }

    return RecordId();
}

Status IndexAccessMethod::validate(OperationContext* txn,
                                   bool full,
                                   int64_t* numKeys,
                                   ValidateResults* fullResults) {
    long long keys = 0;
    _newInterface->fullValidate(txn, full, &keys, fullResults);
    *numKeys = keys;
    return Status::OK();
}

bool IndexAccessMethod::appendCustomStats(OperationContext* txn,
                                          BSONObjBuilder* output,
                                          double scale) const {
    return _newInterface->appendCustomStats(txn, output, scale);
}

long long IndexAccessMethod::getSpaceUsedBytes(OperationContext* txn) const {
    return _newInterface->getSpaceUsedBytes(txn);
}

Status IndexAccessMethod::validateUpdate(OperationContext* txn,
                                         const BSONObj& from,
                                         const BSONObj& to,
                                         const RecordId& record,
                                         const InsertDeleteOptions& options,
                                         UpdateTicket* ticket,
                                         const MatchExpression* indexFilter) {
    if (indexFilter == NULL || indexFilter->matchesBSON(from))
        getKeys(from, &ticket->oldKeys);
    if (indexFilter == NULL || indexFilter->matchesBSON(to))
        getKeys(to, &ticket->newKeys);
    ticket->loc = record;
    ticket->dupsAllowed = options.dupsAllowed;

    setDifference(ticket->oldKeys, ticket->newKeys, &ticket->removed);
    setDifference(ticket->newKeys, ticket->oldKeys, &ticket->added);

    ticket->_isValid = true;

    return Status::OK();
}

Status IndexAccessMethod::update(OperationContext* txn,
                                 const UpdateTicket& ticket,
                                 int64_t* numUpdated) {
    if (!ticket._isValid) {
        return Status(ErrorCodes::InternalError, "Invalid UpdateTicket in update");
    }

    if (ticket.oldKeys.size() + ticket.added.size() - ticket.removed.size() > 1) {
        _btreeState->setMultikey(txn);
    }

    for (size_t i = 0; i < ticket.removed.size(); ++i) {
        _newInterface->unindex(txn, *ticket.removed[i], ticket.loc, ticket.dupsAllowed);
    }

    for (size_t i = 0; i < ticket.added.size(); ++i) {
        Status status =
            _newInterface->insert(txn, *ticket.added[i], ticket.loc, ticket.dupsAllowed);
        if (!status.isOK()) {
            if (status.code() == ErrorCodes::KeyTooLong && ignoreKeyTooLong(txn)) {
                // Ignore.
                continue;
            }

            return status;
        }
    }

    *numUpdated = ticket.added.size();

    return Status::OK();
}

Status IndexAccessMethod::compact(OperationContext* txn) {
    return this->_newInterface->compact(txn);
}

std::unique_ptr<IndexAccessMethod::BulkBuilder> IndexAccessMethod::initiateBulk() {
    return std::unique_ptr<BulkBuilder>(new BulkBuilder(this, _descriptor));
}

IndexAccessMethod::BulkBuilder::BulkBuilder(const IndexAccessMethod* index,
                                            const IndexDescriptor* descriptor)
    : _sorter(Sorter::make(
          SortOptions()
              .TempDir(storageGlobalParams.dbpath + "/_tmp")
              .ExtSortAllowed()
              .MaxMemoryUsageBytes(100 * 1024 * 1024),
          BtreeExternalSortComparison(descriptor->keyPattern(), descriptor->version()))),
      _real(index) {}

Status IndexAccessMethod::BulkBuilder::insert(OperationContext* txn,
                                              const BSONObj& obj,
                                              const RecordId& loc,
                                              const InsertDeleteOptions& options,
                                              int64_t* numInserted) {
    BSONObjSet keys;
    _real->getKeys(obj, &keys);

    _isMultiKey = _isMultiKey || (keys.size() > 1);

    for (BSONObjSet::iterator it = keys.begin(); it != keys.end(); ++it) {
        _sorter->add(*it, loc);
        _keysInserted++;
    }

    if (NULL != numInserted) {
        *numInserted += keys.size();
    }

    return Status::OK();
}


Status IndexAccessMethod::commitBulk(OperationContext* txn,
                                     std::unique_ptr<BulkBuilder> bulk,
                                     bool mayInterrupt,
                                     bool dupsAllowed,
                                     set<RecordId>* dupsToDrop) {
    Timer timer;

    std::unique_ptr<BulkBuilder::Sorter::Iterator> i(bulk->_sorter->done());

    stdx::unique_lock<Client> lk(*txn->getClient());
    ProgressMeterHolder pm(*txn->setMessage_inlock("Index Bulk Build: (2/3) btree bottom up",
                                                   "Index: (2/3) BTree Bottom Up Progress",
                                                   bulk->_keysInserted,
                                                   10));
    lk.unlock();

    std::unique_ptr<SortedDataBuilderInterface> builder;

    MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
        WriteUnitOfWork wunit(txn);

        if (bulk->_isMultiKey) {
            _btreeState->setMultikey(txn);
        }

        builder.reset(_newInterface->getBulkBuilder(txn, dupsAllowed));
        wunit.commit();
    }
    MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "setting index multikey flag", "");

    while (i->more()) {
        if (mayInterrupt) {
            txn->checkForInterrupt();
        }

        WriteUnitOfWork wunit(txn);
        // Improve performance in the btree-building phase by disabling rollback tracking.
        // This avoids copying all the written bytes to a buffer that is only used to roll back.
        // Note that this is safe to do, as this entire index-build-in-progress will be cleaned
        // up by the index system.
        txn->recoveryUnit()->setRollbackWritesDisabled();

        // Get the next datum and add it to the builder.
        BulkBuilder::Sorter::Data d = i->next();
        Status status = builder->addKey(d.first, d.second);

        if (!status.isOK()) {
            // Overlong key that's OK to skip?
            if (status.code() == ErrorCodes::KeyTooLong && ignoreKeyTooLong(txn)) {
                continue;
            }

            // Check if this is a duplicate that's OK to skip
            if (status.code() == ErrorCodes::DuplicateKey) {
                invariant(!dupsAllowed);  // shouldn't be getting DupKey errors if dupsAllowed.

                if (dupsToDrop) {
                    dupsToDrop->insert(d.second);
                    continue;
                }
            }

            return status;
        }

        // If we're here either it's a dup and we're cool with it or the addKey went just
        // fine.
        pm.hit();
        wunit.commit();
    }

    pm.finished();

    {
        stdx::lock_guard<Client> lk(*txn->getClient());
        CurOp::get(txn)->setMessage_inlock("Index Bulk Build: (3/3) btree-middle",
                                           "Index: (3/3) BTree Middle Progress");
    }

    LOG(timer.seconds() > 10 ? 0 : 1) << "\t done building bottom layer, going to commit";

    builder->commit(mayInterrupt);
    return Status::OK();
}

}  // namespace mongo

#include "mongo/db/sorter/sorter.cpp"
MONGO_CREATE_SORTER(mongo::BSONObj, mongo::RecordId, mongo::BtreeExternalSortComparison);