Hawkin
/
arangodb
mirror of https://gitee.com/bigwinds/arangodb
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
arangodb/arangod/RocksDBEngine/RocksDBPrimaryIndex.cpp

1016 lines
35 KiB
C++
Raw Blame History

////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2017 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed 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.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Jan Steemann
////////////////////////////////////////////////////////////////////////////////
#include "Aql/Ast.h"
#include "Aql/AstNode.h"
#include "Basics/Exceptions.h"
#include "Basics/StaticStrings.h"
#include "Basics/VelocyPackHelper.h"
#include "Cache/CachedValue.h"
#include "Cache/TransactionalCache.h"
#include "Cluster/ServerState.h"
#include "Indexes/SortedIndexAttributeMatcher.h"
#include "Logger/Logger.h"
#include "RocksDBEngine/RocksDBCollection.h"
#include "RocksDBEngine/RocksDBCommon.h"
#include "RocksDBEngine/RocksDBComparator.h"
#include "RocksDBEngine/RocksDBEngine.h"
#include "RocksDBEngine/RocksDBKey.h"
#include "RocksDBEngine/RocksDBKeyBounds.h"
#include "RocksDBEngine/RocksDBMethods.h"
#include "RocksDBEngine/RocksDBTransactionState.h"
#include "RocksDBEngine/RocksDBTypes.h"
#include "RocksDBEngine/RocksDBValue.h"
#include "RocksDBPrimaryIndex.h"
#include "StorageEngine/EngineSelectorFeature.h"
#include "Transaction/Context.h"
#include "Transaction/Helpers.h"
#include "Transaction/Methods.h"
#include "Utils/CollectionNameResolver.h"
#include "VocBase/LogicalCollection.h"
#include "RocksDBEngine/RocksDBPrefixExtractor.h"
#ifdef USE_ENTERPRISE
#include "Enterprise/VocBase/VirtualCollection.h"
#endif
#include <rocksdb/iterator.h>
#include <rocksdb/utilities/transaction.h>
#include <velocypack/Builder.h>
#include <velocypack/Collection.h>
#include <velocypack/Slice.h>
#include <velocypack/velocypack-aliases.h>
using namespace arangodb;
namespace {
std::string const lowest; // smallest possible key
std::string const highest = "\xFF"; // greatest possible key
} // namespace
// ================ Primary Index Iterators ================
namespace arangodb {
class RocksDBPrimaryIndexEqIterator final : public IndexIterator {
public:
RocksDBPrimaryIndexEqIterator(LogicalCollection* collection,
transaction::Methods* trx, RocksDBPrimaryIndex* index,
std::unique_ptr<VPackBuilder> key,
bool allowCoveringIndexOptimization)
: IndexIterator(collection, trx),
_index(index),
_key(std::move(key)),
_done(false),
_allowCoveringIndexOptimization(allowCoveringIndexOptimization) {
TRI_ASSERT(_key->slice().isString());
}
~RocksDBPrimaryIndexEqIterator() {
if (_key != nullptr) {
// return the VPackBuilder to the transaction context
_trx->transactionContextPtr()->returnBuilder(_key.release());
}
}
char const* typeName() const override { return "primary-index-eq-iterator"; }
/// @brief index supports rearming
bool canRearm() const override { return true; }
/// @brief rearm the index iterator
bool rearm(arangodb::aql::AstNode const* node, arangodb::aql::Variable const* variable,
IndexIteratorOptions const& opts) override {
TRI_ASSERT(node != nullptr);
TRI_ASSERT(node->type == aql::NODE_TYPE_OPERATOR_NARY_AND);
TRI_ASSERT(node->numMembers() == 1);
AttributeAccessParts aap(node->getMember(0), variable);
TRI_ASSERT(aap.opType == arangodb::aql::NODE_TYPE_OPERATOR_BINARY_EQ);
// handle the sole element
_key->clear();
_index->handleValNode(_trx, _key.get(), aap.value, !_allowCoveringIndexOptimization);
TRI_IF_FAILURE("PrimaryIndex::noIterator") {
THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
}
return !_key->isEmpty();
}
bool next(LocalDocumentIdCallback const& cb, size_t limit) override {
if (limit == 0 || _done) {
// No limit no data, or we are actually done. The last call should have
// returned false
TRI_ASSERT(limit > 0); // Someone called with limit == 0. Api broken
return false;
}
_done = true;
LocalDocumentId documentId =
_index->lookupKey(_trx, arangodb::velocypack::StringRef(_key->slice()));
if (documentId.isSet()) {
cb(documentId);
}
return false;
}
/// @brief extracts just _key. not supported for use with _id
bool nextCovering(DocumentCallback const& cb, size_t limit) override {
TRI_ASSERT(_allowCoveringIndexOptimization);
if (limit == 0 || _done) {
// No limit no data, or we are actually done. The last call should have
// returned false
TRI_ASSERT(limit > 0); // Someone called with limit == 0. Api broken
return false;
}
_done = true;
LocalDocumentId documentId =
_index->lookupKey(_trx, arangodb::velocypack::StringRef(_key->slice()));
if (documentId.isSet()) {
cb(documentId, _key->slice());
}
return false;
}
void reset() override { _done = false; }
/// @brief we provide a method to provide the index attribute values
/// while scanning the index
bool hasCovering() const override { return _allowCoveringIndexOptimization; }
private:
RocksDBPrimaryIndex* _index;
std::unique_ptr<VPackBuilder> _key;
bool _done;
bool const _allowCoveringIndexOptimization;
};
class RocksDBPrimaryIndexInIterator final : public IndexIterator {
public:
RocksDBPrimaryIndexInIterator(LogicalCollection* collection,
transaction::Methods* trx, RocksDBPrimaryIndex* index,
std::unique_ptr<VPackBuilder> keys,
bool allowCoveringIndexOptimization)
: IndexIterator(collection, trx),
_index(index),
_keys(std::move(keys)),
_iterator(_keys->slice()),
_allowCoveringIndexOptimization(allowCoveringIndexOptimization) {
TRI_ASSERT(_keys->slice().isArray());
}
~RocksDBPrimaryIndexInIterator() {
if (_keys != nullptr) {
// return the VPackBuilder to the transaction context
_trx->transactionContextPtr()->returnBuilder(_keys.release());
}
}
char const* typeName() const override { return "primary-index-in-iterator"; }
/// @brief index supports rearming
bool canRearm() const override { return true; }
/// @brief rearm the index iterator
bool rearm(arangodb::aql::AstNode const* node, arangodb::aql::Variable const* variable,
IndexIteratorOptions const& opts) override {
TRI_ASSERT(node != nullptr);
TRI_ASSERT(node->type == aql::NODE_TYPE_OPERATOR_NARY_AND);
TRI_ASSERT(node->numMembers() == 1);
AttributeAccessParts aap(node->getMember(0), variable);
TRI_ASSERT(aap.opType == arangodb::aql::NODE_TYPE_OPERATOR_BINARY_IN);
if (aap.value->isArray()) {
_index->fillInLookupValues(_trx, *(_keys.get()), aap.value, opts.ascending,
!_allowCoveringIndexOptimization);
_iterator = VPackArrayIterator(_keys->slice());
return true;
}
return false;
}
bool next(LocalDocumentIdCallback const& cb, size_t limit) override {
if (limit == 0 || !_iterator.valid()) {
// No limit no data, or we are actually done. The last call should have
// returned false
TRI_ASSERT(limit > 0); // Someone called with limit == 0. Api broken
return false;
}
while (limit > 0) {
LocalDocumentId documentId =
_index->lookupKey(_trx, arangodb::velocypack::StringRef(*_iterator));
if (documentId.isSet()) {
cb(documentId);
--limit;
}
_iterator.next();
if (!_iterator.valid()) {
return false;
}
}
return true;
}
bool nextCovering(DocumentCallback const& cb, size_t limit) override {
TRI_ASSERT(_allowCoveringIndexOptimization);
if (limit == 0 || !_iterator.valid()) {
// No limit no data, or we are actually done. The last call should have
// returned false
TRI_ASSERT(limit > 0); // Someone called with limit == 0. Api broken
return false;
}
while (limit > 0) {
LocalDocumentId documentId =
_index->lookupKey(_trx, arangodb::velocypack::StringRef(*_iterator));
if (documentId.isSet()) {
cb(documentId, *_iterator);
--limit;
}
_iterator.next();
if (!_iterator.valid()) {
return false;
}
}
return true;
}
void reset() override { _iterator.reset(); }
/// @brief we provide a method to provide the index attribute values
/// while scanning the index
bool hasCovering() const override { return _allowCoveringIndexOptimization; }
private:
RocksDBPrimaryIndex* _index;
std::unique_ptr<VPackBuilder> _keys;
arangodb::velocypack::ArrayIterator _iterator;
bool const _allowCoveringIndexOptimization;
};
class RocksDBPrimaryIndexRangeIterator final : public IndexIterator {
private:
friend class RocksDBVPackIndex;
public:
RocksDBPrimaryIndexRangeIterator(LogicalCollection* collection, transaction::Methods* trx,
arangodb::RocksDBPrimaryIndex const* index,
bool reverse, RocksDBKeyBounds&& bounds,
bool allowCoveringIndexOptimization)
: IndexIterator(collection, trx),
_index(index),
_cmp(index->comparator()),
_reverse(reverse),
_allowCoveringIndexOptimization(allowCoveringIndexOptimization),
_bounds(std::move(bounds)) {
TRI_ASSERT(index->columnFamily() == RocksDBColumnFamily::primary());
RocksDBMethods* mthds = RocksDBTransactionState::toMethods(trx);
rocksdb::ReadOptions options = mthds->iteratorReadOptions();
// we need to have a pointer to a slice for the upper bound
// so we need to assign the slice to an instance variable here
if (reverse) {
_rangeBound = _bounds.start();
options.iterate_lower_bound = &_rangeBound;
} else {
_rangeBound = _bounds.end();
options.iterate_upper_bound = &_rangeBound;
}
TRI_ASSERT(options.prefix_same_as_start);
_iterator = mthds->NewIterator(options, index->columnFamily());
if (reverse) {
_iterator->SeekForPrev(_bounds.end());
} else {
_iterator->Seek(_bounds.start());
}
}
public:
char const* typeName() const override {
return "rocksdb-range-index-iterator";
}
/// @brief Get the next limit many elements in the index
bool next(LocalDocumentIdCallback const& cb, size_t limit) override {
TRI_ASSERT(_trx->state()->isRunning());
if (limit == 0 || !_iterator->Valid() || outOfRange()) {
// No limit no data, or we are actually done. The last call should have
// returned false
TRI_ASSERT(limit > 0); // Someone called with limit == 0. Api broken
return false;
}
while (limit > 0) {
TRI_ASSERT(_index->objectId() == RocksDBKey::objectId(_iterator->key()));
cb(RocksDBValue::documentId(_iterator->value()));
--limit;
if (_reverse) {
_iterator->Prev();
} else {
_iterator->Next();
}
if (!_iterator->Valid() || outOfRange()) {
return false;
}
}
return true;
}
bool nextCovering(DocumentCallback const& cb, size_t limit) override {
TRI_ASSERT(_allowCoveringIndexOptimization);
if (limit == 0 || !_iterator->Valid() || outOfRange()) {
// No limit no data, or we are actually done. The last call should have
// returned false
TRI_ASSERT(limit > 0); // Someone called with limit == 0. Api broken
return false;
}
transaction::BuilderLeaser builder(transaction());
while (limit > 0) {
LocalDocumentId documentId = RocksDBValue::documentId(_iterator->value());
arangodb::velocypack::StringRef key = RocksDBKey::primaryKey(_iterator->key());
builder->clear();
builder->add(VPackValuePair(key.data(), key.size(), VPackValueType::String));
cb(documentId, builder->slice());
--limit;
if (_reverse) {
_iterator->Prev();
} else {
_iterator->Next();
}
if (!_iterator->Valid() || outOfRange()) {
return false;
}
}
return true;
}
void skip(uint64_t count, uint64_t& skipped) override {
TRI_ASSERT(_trx->state()->isRunning());
if (!_iterator->Valid() || outOfRange()) {
return;
}
while (count > 0) {
TRI_ASSERT(_index->objectId() == RocksDBKey::objectId(_iterator->key()));
--count;
++skipped;
if (_reverse) {
_iterator->Prev();
} else {
_iterator->Next();
}
if (!_iterator->Valid() || outOfRange()) {
return;
}
}
}
/// @brief Reset the cursor
void reset() override {
TRI_ASSERT(_trx->state()->isRunning());
if (_reverse) {
_iterator->SeekForPrev(_bounds.end());
} else {
_iterator->Seek(_bounds.start());
}
}
/// @brief we provide a method to provide the index attribute values
/// while scanning the index
bool hasCovering() const override { return _allowCoveringIndexOptimization; }
private:
bool outOfRange() const {
TRI_ASSERT(_trx->state()->isRunning());
if (_reverse) {
return (_cmp->Compare(_iterator->key(), _bounds.start()) < 0);
} else {
return (_cmp->Compare(_iterator->key(), _bounds.end()) > 0);
}
}
arangodb::RocksDBPrimaryIndex const* _index;
rocksdb::Comparator const* _cmp;
std::unique_ptr<rocksdb::Iterator> _iterator;
bool const _reverse;
bool const _allowCoveringIndexOptimization;
RocksDBKeyBounds _bounds;
// used for iterate_upper_bound iterate_lower_bound
rocksdb::Slice _rangeBound;
};
} // namespace arangodb
// ================ PrimaryIndex ================
RocksDBPrimaryIndex::RocksDBPrimaryIndex(arangodb::LogicalCollection& collection,
arangodb::velocypack::Slice const& info)
: RocksDBIndex(0, collection, StaticStrings::IndexNamePrimary,
std::vector<std::vector<arangodb::basics::AttributeName>>(
{{arangodb::basics::AttributeName(StaticStrings::KeyString, false)}}),
true, false, RocksDBColumnFamily::primary(),
basics::VelocyPackHelper::stringUInt64(info, "objectId"),
static_cast<RocksDBCollection*>(collection.getPhysical())->cacheEnabled()),
_isRunningInCluster(ServerState::instance()->isRunningInCluster()) {
TRI_ASSERT(_cf == RocksDBColumnFamily::primary());
TRI_ASSERT(_objectId != 0);
}
RocksDBPrimaryIndex::~RocksDBPrimaryIndex() {}
void RocksDBPrimaryIndex::load() {
RocksDBIndex::load();
if (useCache()) {
// FIXME: make the factor configurable
RocksDBCollection* rdb = static_cast<RocksDBCollection*>(_collection.getPhysical());
uint64_t numDocs = rdb->numberDocuments();
if (numDocs > 0) {
_cache->sizeHint(static_cast<uint64_t>(0.3 * numDocs));
}
}
}
/// @brief return a VelocyPack representation of the index
void RocksDBPrimaryIndex::toVelocyPack(VPackBuilder& builder,
std::underlying_type<Serialize>::type flags) const {
builder.openObject();
RocksDBIndex::toVelocyPack(builder, flags);
// hard-coded
builder.add(arangodb::StaticStrings::IndexUnique, arangodb::velocypack::Value(true));
builder.add(arangodb::StaticStrings::IndexSparse, arangodb::velocypack::Value(false));
builder.close();
}
LocalDocumentId RocksDBPrimaryIndex::lookupKey(transaction::Methods* trx,
arangodb::velocypack::StringRef keyRef) const {
RocksDBKeyLeaser key(trx);
key->constructPrimaryIndexValue(_objectId, keyRef);
bool lockTimeout = false;
if (useCache()) {
TRI_ASSERT(_cache != nullptr);
// check cache first for fast path
auto f = _cache->find(key->string().data(),
static_cast<uint32_t>(key->string().size()));
if (f.found()) {
rocksdb::Slice s(reinterpret_cast<char const*>(f.value()->value()),
f.value()->valueSize());
return RocksDBValue::documentId(s);
} else if (f.result().errorNumber() == TRI_ERROR_LOCK_TIMEOUT) {
// assuming someone is currently holding a write lock, which
// is why we cannot access the TransactionalBucket.
lockTimeout = true; // we skip the insert in this case
}
}
RocksDBMethods* mthds = RocksDBTransactionState::toMethods(trx);
rocksdb::PinnableSlice val;
rocksdb::Status s = mthds->Get(_cf, key->string(), &val);
if (!s.ok()) {
return LocalDocumentId();
}
if (useCache() && !lockTimeout) {
TRI_ASSERT(_cache != nullptr);
// write entry back to cache
auto entry =
cache::CachedValue::construct(key->string().data(),
static_cast<uint32_t>(key->string().size()),
val.data(), static_cast<uint64_t>(val.size()));
if (entry) {
Result status = _cache->insert(entry);
if (status.errorNumber() == TRI_ERROR_LOCK_TIMEOUT) {
// the writeLock uses cpu_relax internally, so we can try yield
std::this_thread::yield();
status = _cache->insert(entry);
}
if (status.fail()) {
delete entry;
}
}
}
return RocksDBValue::documentId(val);
}
/// @brief reads a revision id from the primary index
/// if the document does not exist, this function will return false
/// if the document exists, the function will return true
/// the revision id will only be non-zero if the primary index
/// value contains the document's revision id. note that this is not
/// the case for older collections
/// in this case the caller must fetch the revision id from the actual
/// document
bool RocksDBPrimaryIndex::lookupRevision(transaction::Methods* trx,
arangodb::velocypack::StringRef keyRef,
LocalDocumentId& documentId,
TRI_voc_rid_t& revisionId) const {
documentId.clear();
revisionId = 0;
RocksDBKeyLeaser key(trx);
key->constructPrimaryIndexValue(_objectId, keyRef);
// acquire rocksdb transaction
RocksDBMethods* mthds = RocksDBTransactionState::toMethods(trx);
rocksdb::PinnableSlice val;
rocksdb::Status s = mthds->Get(_cf, key->string(), &val);
if (!s.ok()) {
return false;
}
documentId = RocksDBValue::documentId(val);
// this call will populate revisionId if the revision id value is
// stored in the primary index
revisionId = RocksDBValue::revisionId(val);
return true;
}
Result RocksDBPrimaryIndex::insert(transaction::Methods& trx, RocksDBMethods* mthd,
LocalDocumentId const& documentId,
velocypack::Slice const& slice,
Index::OperationMode mode) {
VPackSlice keySlice;
TRI_voc_rid_t revision;
transaction::helpers::extractKeyAndRevFromDocument(slice, keySlice, revision);
TRI_ASSERT(keySlice.isString());
RocksDBKeyLeaser key(&trx);
key->constructPrimaryIndexValue(_objectId, arangodb::velocypack::StringRef(keySlice));
transaction::StringLeaser leased(&trx);
rocksdb::PinnableSlice ps(leased.get());
rocksdb::Status s = mthd->GetForUpdate(_cf, key->string(), &ps);
Result res;
if (s.ok()) { // detected conflicting primary key
std::string existingId = keySlice.copyString();
if (mode == OperationMode::internal) {
return res.reset(TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED, std::move(existingId));
}
res.reset(TRI_ERROR_ARANGO_UNIQUE_CONSTRAINT_VIOLATED);
return addErrorMsg(res, existingId);
} else if (!s.IsNotFound()) {
// IsBusy(), IsTimedOut() etc... this indicates a conflict
return addErrorMsg(res.reset(rocksutils::convertStatus(s)));
}
ps.Reset(); // clear used memory
if (trx.state()->hasHint(transaction::Hints::Hint::GLOBAL_MANAGED)) {
// blacklist new index entry to avoid caching without committing first
blackListKey(key->string().data(), static_cast<uint32_t>(key->string().size()));
}
auto value = RocksDBValue::PrimaryIndexValue(documentId, revision);
s = mthd->Put(_cf, key.ref(), value.string(), /*assume_tracked*/true);
if (!s.ok()) {
res.reset(rocksutils::convertStatus(s, rocksutils::index));
addErrorMsg(res);
}
return res;
}
Result RocksDBPrimaryIndex::update(transaction::Methods& trx, RocksDBMethods* mthd,
LocalDocumentId const& oldDocumentId,
velocypack::Slice const& oldDoc,
LocalDocumentId const& newDocumentId,
velocypack::Slice const& newDoc,
Index::OperationMode mode) {
Result res;
VPackSlice keySlice = transaction::helpers::extractKeyFromDocument(oldDoc);
TRI_ASSERT(keySlice.binaryEquals(oldDoc.get(StaticStrings::KeyString)));
RocksDBKeyLeaser key(&trx);
key->constructPrimaryIndexValue(_objectId, arangodb::velocypack::StringRef(keySlice));
TRI_voc_rid_t revision = transaction::helpers::extractRevFromDocument(newDoc);
auto value = RocksDBValue::PrimaryIndexValue(newDocumentId, revision);
// blacklist new index entry to avoid caching without committing first
blackListKey(key->string().data(), static_cast<uint32_t>(key->string().size()));
rocksdb::Status s = mthd->Put(_cf, key.ref(), value.string(), /*assume_tracked*/false);
if (!s.ok()) {
res.reset(rocksutils::convertStatus(s, rocksutils::index));
addErrorMsg(res);
}
return res;
}
Result RocksDBPrimaryIndex::remove(transaction::Methods& trx, RocksDBMethods* mthd,
LocalDocumentId const& documentId,
velocypack::Slice const& slice,
Index::OperationMode mode) {
Result res;
// TODO: deal with matching revisions?
VPackSlice keySlice = transaction::helpers::extractKeyFromDocument(slice);
TRI_ASSERT(keySlice.isString());
RocksDBKeyLeaser key(&trx);
key->constructPrimaryIndexValue(_objectId, arangodb::velocypack::StringRef(keySlice));
blackListKey(key->string().data(), static_cast<uint32_t>(key->string().size()));
// acquire rocksdb transaction
auto* mthds = RocksDBTransactionState::toMethods(&trx);
rocksdb::Status s = mthds->Delete(_cf, key.ref());
if (!s.ok()) {
res.reset(rocksutils::convertStatus(s, rocksutils::index));
addErrorMsg(res);
}
return res;
}
/// @brief checks whether the index supports the condition
Index::FilterCosts RocksDBPrimaryIndex::supportsFilterCondition(
std::vector<std::shared_ptr<arangodb::Index>> const& allIndexes,
arangodb::aql::AstNode const* node, arangodb::aql::Variable const* reference,
size_t itemsInIndex) const {
return SortedIndexAttributeMatcher::supportsFilterCondition(allIndexes, this, node, reference, itemsInIndex);
}
Index::SortCosts RocksDBPrimaryIndex::supportsSortCondition(arangodb::aql::SortCondition const* sortCondition,
arangodb::aql::Variable const* reference,
size_t itemsInIndex) const {
return SortedIndexAttributeMatcher::supportsSortCondition(this, sortCondition, reference, itemsInIndex);
}
/// @brief creates an IndexIterator for the given Condition
std::unique_ptr<IndexIterator> RocksDBPrimaryIndex::iteratorForCondition(
transaction::Methods* trx, arangodb::aql::AstNode const* node,
arangodb::aql::Variable const* reference, IndexIteratorOptions const& opts) {
TRI_ASSERT(!isSorted() || opts.sorted);
if (node == nullptr) {
// full range scan
return std::make_unique<RocksDBPrimaryIndexRangeIterator>(
&_collection /*logical collection*/, trx, this, !opts.ascending /*reverse*/,
RocksDBKeyBounds::PrimaryIndex(_objectId, ::lowest, ::highest), opts.forceProjection);
}
TRI_ASSERT(node != nullptr);
TRI_ASSERT(node->type == aql::NODE_TYPE_OPERATOR_NARY_AND);
size_t const n = node->numMembers();
TRI_ASSERT(n >= 1);
if (n == 1) {
AttributeAccessParts aap(node->getMember(0), reference);
if (aap.opType == aql::NODE_TYPE_OPERATOR_BINARY_EQ) {
// a.b == value
return createEqIterator(trx, aap.attribute, aap.value);
}
if (aap.opType == aql::NODE_TYPE_OPERATOR_BINARY_IN && aap.value->isArray()) {
// a.b IN array
return createInIterator(trx, aap.attribute, aap.value, opts.ascending);
}
// fall-through intentional here
}
auto removeCollectionFromString = [this, &trx](bool isId, std::string& value) -> int {
if (isId) {
char const* key = nullptr;
size_t outLength = 0;
std::shared_ptr<LogicalCollection> collection;
Result res = trx->resolveId(value.data(), value.length(), collection, key, outLength);
if (!res.ok()) {
// using the name of an unknown collection
if (_isRunningInCluster) {
// translate from our own shard name to "real" collection name
return value.compare(trx->resolver()->getCollectionName(_collection.id()));
}
return value.compare(_collection.name());
}
TRI_ASSERT(key);
TRI_ASSERT(collection);
if (!_isRunningInCluster && collection->id() != _collection.id()) {
// using the name of a different collection...
return value.compare(_collection.name());
} else if (_isRunningInCluster && collection->planId() != _collection.planId()) {
// using a different collection
// translate from our own shard name to "real" collection name
return value.compare(trx->resolver()->getCollectionName(_collection.id()));
}
// strip collection name prefix
value = std::string(key, outLength);
}
// usage of _key or same collection name
return 0;
};
std::string lower;
std::string upper;
bool lowerFound = false;
bool upperFound = false;
for (size_t i = 0; i < n; ++i) {
AttributeAccessParts aap(node->getMemberUnchecked(i), reference);
auto type = aap.opType;
if (!(type == aql::NODE_TYPE_OPERATOR_BINARY_LE || type == aql::NODE_TYPE_OPERATOR_BINARY_LT ||
type == aql::NODE_TYPE_OPERATOR_BINARY_GE || type == aql::NODE_TYPE_OPERATOR_BINARY_GT ||
type == aql::NODE_TYPE_OPERATOR_BINARY_EQ)) {
return std::make_unique<EmptyIndexIterator>(&_collection, trx);
}
TRI_ASSERT(aap.attribute->type == aql::NODE_TYPE_ATTRIBUTE_ACCESS);
bool const isId = (aap.attribute->stringEquals(StaticStrings::IdString));
std::string value; // empty string == lower bound
if (aap.value->isStringValue()) {
value = aap.value->getString();
} else if (aap.value->isObject() || aap.value->isArray()) {
// any array or object value is bigger than any potential key
value = ::highest;
} else if (aap.value->isNullValue() || aap.value->isBoolValue() ||
aap.value->isIntValue()) {
// any null, bool or numeric value is lower than any potential key
// keep lower bound
} else {
THROW_ARANGO_EXCEPTION_MESSAGE(TRI_ERROR_INTERNAL,
std::string(
"unhandled type for valNode: ") +
aap.value->getTypeString());
}
// strip collection name prefix from comparison value
int const cmpResult = removeCollectionFromString(isId, value);
if (type == aql::NODE_TYPE_OPERATOR_BINARY_EQ) {
if (cmpResult != 0) {
// doc._id == different collection
return std::make_unique<EmptyIndexIterator>(&_collection, trx);
}
if (!upperFound || value < upper) {
upper = value;
upperFound = true;
}
if (!lowerFound || value < lower) {
lower = std::move(value);
lowerFound = true;
}
} else if (type == aql::NODE_TYPE_OPERATOR_BINARY_LE ||
type == aql::NODE_TYPE_OPERATOR_BINARY_LT) {
// a.b < value
if (cmpResult > 0) {
// doc._id < collection with "bigger" name
upper = ::highest;
} else if (cmpResult < 0) {
// doc._id < collection with "lower" name
return std::make_unique<EmptyIndexIterator>(&_collection, trx);
} else {
if (type == aql::NODE_TYPE_OPERATOR_BINARY_LT && !value.empty()) {
value.back() -= 0x01U; // modify upper bound so that it is not included
}
if (!upperFound || value < upper) {
upper = std::move(value);
}
}
upperFound = true;
} else if (type == aql::NODE_TYPE_OPERATOR_BINARY_GE ||
type == aql::NODE_TYPE_OPERATOR_BINARY_GT) {
// a.b > value
if (cmpResult < 0) {
// doc._id > collection with "smaller" name
lower = ::lowest;
} else if (cmpResult > 0) {
// doc._id > collection with "bigger" name
return std::make_unique<EmptyIndexIterator>(&_collection, trx);
} else {
if (type == aql::NODE_TYPE_OPERATOR_BINARY_GE && !value.empty()) {
value.back() -= 0x01U; // modify lower bound so it is included
}
if (!lowerFound || value > lower) {
lower = std::move(value);
}
}
lowerFound = true;
}
} // for nodes
// if only one bound is given select the other (lowest or highest) accordingly
if (upperFound && !lowerFound) {
lower = ::lowest;
lowerFound = true;
} else if (lowerFound && !upperFound) {
upper = ::highest;
upperFound = true;
}
if (lowerFound && upperFound) {
return std::make_unique<RocksDBPrimaryIndexRangeIterator>(
&_collection /*logical collection*/, trx, this, !opts.ascending /*reverse*/,
RocksDBKeyBounds::PrimaryIndex(_objectId, lower, upper), opts.forceProjection);
}
// operator type unsupported or IN used on non-array
return std::make_unique<EmptyIndexIterator>(&_collection, trx);
}
/// @brief specializes the condition for use with the index
arangodb::aql::AstNode* RocksDBPrimaryIndex::specializeCondition(
arangodb::aql::AstNode* node, arangodb::aql::Variable const* reference) const {
return SortedIndexAttributeMatcher::specializeCondition(this, node, reference);
}
/// @brief create the iterator, for a single attribute, IN operator
std::unique_ptr<IndexIterator> RocksDBPrimaryIndex::createInIterator(transaction::Methods* trx,
arangodb::aql::AstNode const* attrNode,
arangodb::aql::AstNode const* valNode,
bool ascending) {
// _key or _id?
bool const isId = (attrNode->stringEquals(StaticStrings::IdString));
TRI_ASSERT(valNode->isArray());
// lease builder, but immediately pass it to the unique_ptr so we don't leak
transaction::BuilderLeaser builder(trx);
std::unique_ptr<VPackBuilder> keys(builder.steal());
fillInLookupValues(trx, *(keys.get()), valNode, ascending, isId);
return std::make_unique<RocksDBPrimaryIndexInIterator>(&_collection, trx, this, std::move(keys), !isId);
}
/// @brief create the iterator, for a single attribute, EQ operator
std::unique_ptr<IndexIterator> RocksDBPrimaryIndex::createEqIterator(transaction::Methods* trx,
arangodb::aql::AstNode const* attrNode,
arangodb::aql::AstNode const* valNode) {
// _key or _id?
bool const isId = (attrNode->stringEquals(StaticStrings::IdString));
// lease builder, but immediately pass it to the unique_ptr so we don't leak
transaction::BuilderLeaser builder(trx);
std::unique_ptr<VPackBuilder> key(builder.steal());
// handle the sole element
handleValNode(trx, key.get(), valNode, isId);
TRI_IF_FAILURE("PrimaryIndex::noIterator") {
THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
}
if (!key->isEmpty()) {
return std::make_unique<RocksDBPrimaryIndexEqIterator>(&_collection, trx, this, std::move(key), !isId);
}
return std::make_unique<EmptyIndexIterator>(&_collection, trx);
}
void RocksDBPrimaryIndex::fillInLookupValues(transaction::Methods* trx, VPackBuilder& keys,
arangodb::aql::AstNode const* values,
bool ascending, bool isId) const {
TRI_ASSERT(values != nullptr);
TRI_ASSERT(values->type == arangodb::aql::NODE_TYPE_ARRAY);
keys.clear();
keys.openArray();
size_t const n = values->numMembers();
// only leave the valid elements
if (ascending) {
for (size_t i = 0; i < n; ++i) {
handleValNode(trx, &keys, values->getMemberUnchecked(i), isId);
TRI_IF_FAILURE("PrimaryIndex::iteratorValNodes") {
THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
}
}
} else {
size_t i = n;
while (i > 0) {
--i;
handleValNode(trx, &keys, values->getMemberUnchecked(i), isId);
TRI_IF_FAILURE("PrimaryIndex::iteratorValNodes") {
THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
}
}
}
TRI_IF_FAILURE("PrimaryIndex::noIterator") {
THROW_ARANGO_EXCEPTION(TRI_ERROR_DEBUG);
}
keys.close();
}
/// @brief add a single value node to the iterator's keys
void RocksDBPrimaryIndex::handleValNode(transaction::Methods* trx, VPackBuilder* keys,
arangodb::aql::AstNode const* valNode,
bool isId) const {
if (!valNode->isStringValue() || valNode->getStringLength() == 0) {
return;
}
if (isId) {
// lookup by _id. now validate if the lookup is performed for the
// correct collection (i.e. _collection)
char const* key = nullptr;
size_t outLength = 0;
std::shared_ptr<LogicalCollection> collection;
Result res = trx->resolveId(valNode->getStringValue(), valNode->getStringLength(),
collection, key, outLength);
if (!res.ok()) {
return;
}
TRI_ASSERT(collection != nullptr);
TRI_ASSERT(key != nullptr);
if (!_isRunningInCluster && collection->id() != _collection.id()) {
// only continue lookup if the id value is syntactically correct and
// refers to "our" collection, using local collection id
return;
}
if (_isRunningInCluster) {
#ifdef USE_ENTERPRISE
if (collection->isSmart() && collection->type() == TRI_COL_TYPE_EDGE) {
auto c = dynamic_cast<VirtualSmartEdgeCollection const*>(collection.get());
if (c == nullptr) {
THROW_ARANGO_EXCEPTION_MESSAGE(
TRI_ERROR_INTERNAL, "unable to cast smart edge collection");
}
if (_collection.planId() != c->getLocalCid() &&
_collection.planId() != c->getFromCid() &&
_collection.planId() != c->getToCid()) {
// invalid planId
return;
}
} else
#endif
if (collection->planId() != _collection.planId()) {
// only continue lookup if the id value is syntactically correct and
// refers to "our" collection, using cluster collection id
return;
}
}
// use _key value from _id
keys->add(VPackValuePair(key, outLength, VPackValueType::String));
} else {
keys->add(VPackValuePair(valNode->getStringValue(),
valNode->getStringLength(), VPackValueType::String));
}
}
Reference in New Issue View Git Blame Copy Permalink