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

711 lines
25 KiB
C++
Raw Permalink Blame History

////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2017 ArangoDB 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 Simon Grätzer
////////////////////////////////////////////////////////////////////////////////
#include "Basics/Common.h"
#include "Basics/ReadLocker.h"
#include "Basics/StringUtils.h"
#include "Basics/VelocyPackHelper.h"
#include "Basics/conversions.h"
#include "Basics/tri-strings.h"
#include "Cluster/ClusterFeature.h"
#include "Cluster/ClusterInfo.h"
#include "Cluster/ClusterMethods.h"
#include "Cluster/ServerState.h"
#include "GeneralServer/AuthenticationFeature.h"
#include "Indexes.h"
#include "Indexes/Index.h"
#include "Indexes/IndexFactory.h"
#include "Rest/HttpRequest.h"
#include "RestServer/DatabaseFeature.h"
#include "StorageEngine/EngineSelectorFeature.h"
#include "StorageEngine/StorageEngine.h"
#include "Transaction/Helpers.h"
#include "Transaction/Hints.h"
#include "Transaction/StandaloneContext.h"
#include "Transaction/V8Context.h"
#include "Utils/Events.h"
#include "Utils/ExecContext.h"
#include "Utils/SingleCollectionTransaction.h"
#include "V8Server/v8-collection.h"
#include "VocBase/LogicalCollection.h"
#include "VocBase/vocbase.h"
#include "Logger/Logger.h"
#include "Logger/LogMacros.h"
#include <velocypack/Builder.h>
#include <velocypack/Collection.h>
#include <velocypack/Iterator.h>
#include <velocypack/velocypack-aliases.h>
#include <regex>
using namespace arangodb;
using namespace arangodb::basics;
using namespace arangodb::methods;
Result Indexes::getIndex(LogicalCollection const* collection, VPackSlice const& indexId,
VPackBuilder& out, transaction::Methods* trx) {
// do some magic to parse the iid
std::string id; // will (eventually) be fully-qualified; "collection/identifier"
std::string name; // will be just name or id (no "collection/")
VPackSlice idSlice = indexId;
if (idSlice.isObject() && idSlice.hasKey(StaticStrings::IndexId)) {
idSlice = idSlice.get(StaticStrings::IndexId);
}
if (idSlice.isString()) {
std::regex re = std::regex("^([a-zA-Z0-9\\-_]+)\\/([a-zA-Z0-9\\-_]+)$",
std::regex::ECMAScript);
if (std::regex_match(idSlice.copyString(), re)) {
id = idSlice.copyString();
name = id.substr(id.find_first_of("/") + 1);
} else {
name = idSlice.copyString();
id = collection->name() + "/" + name;
}
} else if (idSlice.isInteger()) {
name = StringUtils::itoa(idSlice.getUInt());
id = collection->name() + "/" + name;
} else {
return Result(TRI_ERROR_ARANGO_INDEX_NOT_FOUND);
}
VPackBuilder tmp;
Result res =
Indexes::getAll(collection, Index::makeFlags(), /*withHidden*/ true, tmp, trx);
if (res.ok()) {
for (VPackSlice const& index : VPackArrayIterator(tmp.slice())) {
if (index.get(StaticStrings::IndexId).compareString(id) == 0 ||
index.get(StaticStrings::IndexName).compareString(name) == 0) {
out.add(index);
return Result();
}
}
}
return Result(TRI_ERROR_ARANGO_INDEX_NOT_FOUND);
}
/// @brief get all indexes, skips view links
arangodb::Result Indexes::getAll(LogicalCollection const* collection,
std::underlying_type<Index::Serialize>::type flags,
bool withHidden, VPackBuilder& result,
transaction::Methods* inputTrx) {
VPackBuilder tmp;
if (ServerState::instance()->isCoordinator()) {
TRI_ASSERT(collection);
auto& databaseName = collection->vocbase().name();
std::string const& cid = collection->name();
// add code for estimates here
std::unordered_map<std::string, double> estimates;
int rv = selectivityEstimatesOnCoordinator(databaseName, cid, estimates);
if (rv != TRI_ERROR_NO_ERROR) {
return Result(rv, "could not retrieve estimates");
}
// we will merge in the index estimates later
flags &= ~Index::makeFlags(Index::Serialize::Estimates);
VPackBuilder tmpInner;
auto c = ClusterInfo::instance()->getCollection(databaseName, cid);
c->getIndexesVPack(tmpInner, [withHidden, flags](arangodb::Index const* idx, decltype(flags)& indexFlags) {
if (withHidden || !idx->isHidden()) {
indexFlags = flags;
return true;
}
return false;
});
tmp.openArray();
for (VPackSlice const& s : VPackArrayIterator(tmpInner.slice())) {
auto id = arangodb::velocypack::StringRef(s.get(StaticStrings::IndexId));
auto found = std::find_if(estimates.begin(), estimates.end(),
[&id](std::pair<std::string, double> const& v) {
return id == v.first;
});
if (found == estimates.end()) {
tmp.add(s); // just copy
} else {
tmp.openObject();
tmp.add(VPackObjectIterator(s, true));
tmp.add("selectivityEstimate", VPackValue(found->second));
tmp.close();
}
}
tmp.close();
} else {
std::shared_ptr<transaction::Methods> trx;
if (inputTrx) {
trx = std::shared_ptr<transaction::Methods>(inputTrx, [](transaction::Methods*) {});
} else {
trx = std::make_shared<SingleCollectionTransaction>(
transaction::StandaloneContext::Create(collection->vocbase()),
*collection, AccessMode::Type::READ);
// we actually need this hint here, so that the collection is not
// loaded if it has status unloaded.
trx->addHint(transaction::Hints::Hint::NO_USAGE_LOCK);
Result res = trx->begin();
if (!res.ok()) {
return res;
}
}
// get list of indexes
auto indexes = collection->getIndexes();
tmp.openArray(true);
for (std::shared_ptr<arangodb::Index> const& idx : indexes) {
if (!withHidden && idx->isHidden()) {
continue;
}
idx->toVelocyPack(tmp, flags);
}
tmp.close();
if (!inputTrx) {
Result res;
res = trx->finish(res);
if (res.fail()) {
return res;
}
}
}
bool mergeEdgeIdxs = !ServerState::instance()->isDBServer();
double selectivity = 0, memory = 0, cacheSize = 0, cacheUsage = 0,
cacheLifeTimeHitRate = 0, cacheWindowedHitRate = 0;
VPackArrayBuilder a(&result);
for (VPackSlice const& index : VPackArrayIterator(tmp.slice())) {
std::string id = collection->name() + TRI_INDEX_HANDLE_SEPARATOR_CHR +
index.get(arangodb::StaticStrings::IndexId).copyString();
VPackBuilder merge;
merge.openObject(true);
merge.add(arangodb::StaticStrings::IndexId, arangodb::velocypack::Value(id));
auto type = index.get(arangodb::StaticStrings::IndexType);
if (mergeEdgeIdxs && Index::type(type.copyString()) == Index::TRI_IDX_TYPE_EDGE_INDEX) {
VPackSlice fields = index.get(StaticStrings::IndexFields);
TRI_ASSERT(fields.isArray() && fields.length() <= 2);
if (fields.length() == 1) { // merge indexes
// read out relevant values
VPackSlice val = index.get("selectivityEstimate");
if (val.isNumber()) {
selectivity += val.getNumber<double>();
}
bool useCache = false;
VPackSlice figures = index.get("figures");
if (figures.isObject() && !figures.isEmptyObject()) {
if ((val = figures.get("cacheInUse")).isBool()) {
useCache = val.getBool();
}
if ((val = figures.get("memory")).isNumber()) {
memory += val.getNumber<double>();
}
if ((val = figures.get("cacheSize")).isNumber()) {
cacheSize += val.getNumber<double>();
}
if ((val = figures.get("cacheUsage")).isNumber()) {
cacheUsage += val.getNumber<double>();
}
if ((val = figures.get("cacheLifeTimeHitRate")).isNumber()) {
cacheLifeTimeHitRate += val.getNumber<double>();
}
if ((val = figures.get("cacheWindowedHitRate")).isNumber()) {
cacheWindowedHitRate += val.getNumber<double>();
}
}
if (fields[0].compareString(StaticStrings::FromString) == 0) {
continue;
} else if (fields[0].compareString(StaticStrings::ToString) == 0) {
merge.add(StaticStrings::IndexFields, VPackValue(VPackValueType::Array));
merge.add(VPackValue(StaticStrings::FromString));
merge.add(VPackValue(StaticStrings::ToString));
merge.close();
merge.add("selectivityEstimate", VPackValue(selectivity / 2));
if (Index::hasFlag(flags, Index::Serialize::Figures)) {
merge.add("figures", VPackValue(VPackValueType::Object));
merge.add("memory", VPackValue(memory));
if (useCache) {
merge.add("cacheSize", VPackValue(cacheSize));
merge.add("cacheUsage", VPackValue(cacheUsage));
merge.add("cacheLifeTimeHitRate", VPackValue(cacheLifeTimeHitRate / 2));
merge.add("cacheWindowedHitRate", VPackValue(cacheWindowedHitRate / 2));
}
merge.close();
}
}
}
}
merge.close();
merge = VPackCollection::merge(index, merge.slice(), true);
result.add(merge.slice());
}
return Result();
}
////////////////////////////////////////////////////////////////////////////////
/// @brief ensures an index, locally
////////////////////////////////////////////////////////////////////////////////
static Result EnsureIndexLocal(arangodb::LogicalCollection* collection,
VPackSlice const& definition, bool create,
VPackBuilder& output) {
TRI_ASSERT(collection != nullptr);
Result res;
bool created = false;
std::shared_ptr<arangodb::Index> idx;
READ_LOCKER(readLocker, collection->vocbase()._inventoryLock);
if (create) {
try {
idx = collection->createIndex(definition, created);
} catch (arangodb::basics::Exception const& e) {
return res.reset(e.code(), e.what());
}
TRI_ASSERT(idx != nullptr);
} else {
idx = collection->lookupIndex(definition);
if (idx == nullptr) {
// Index not found
return res.reset(TRI_ERROR_ARANGO_INDEX_NOT_FOUND);
}
}
readLocker.unlock();
TRI_ASSERT(idx != nullptr);
VPackBuilder tmp;
try {
idx->toVelocyPack(tmp, Index::makeFlags(Index::Serialize::Estimates));
} catch (...) {
return res.reset(TRI_ERROR_OUT_OF_MEMORY);
}
std::string iid = StringUtils::itoa(idx->id());
VPackBuilder b;
b.openObject();
b.add("isNewlyCreated", VPackValue(created));
b.add(StaticStrings::IndexId,
VPackValue(collection->name() + TRI_INDEX_HANDLE_SEPARATOR_CHR + iid));
b.close();
output = VPackCollection::merge(tmp.slice(), b.slice(), false);
return res;
}
Result Indexes::ensureIndexCoordinator(arangodb::LogicalCollection const* collection,
VPackSlice const& indexDef, bool create,
VPackBuilder& resultBuilder) {
TRI_ASSERT(collection != nullptr);
auto cluster = application_features::ApplicationServer::getFeature<ClusterFeature>(
"Cluster");
return ClusterInfo::instance()->ensureIndexCoordinator( // create index
*collection, indexDef, create, resultBuilder, cluster->indexCreationTimeout());
}
Result Indexes::ensureIndex(LogicalCollection* collection, VPackSlice const& input,
bool create, VPackBuilder& output) {
// can read indexes with RO on db and collection. Modifications require RW/RW
ExecContext const* exec = ExecContext::CURRENT;
if (exec != nullptr) {
auth::Level lvl = exec->databaseAuthLevel();
bool canModify = exec->canUseCollection(collection->name(), auth::Level::RW);
bool canRead = exec->canUseCollection(collection->name(), auth::Level::RO);
if ((create && (lvl != auth::Level::RW || !canModify)) ||
(lvl == auth::Level::NONE || !canRead)) {
events::CreateIndex(collection->vocbase().name(), collection->name(),
input, TRI_ERROR_FORBIDDEN);
return Result(TRI_ERROR_FORBIDDEN);
}
}
TRI_ASSERT(collection);
VPackBuilder normalized;
StorageEngine* engine = EngineSelectorFeature::ENGINE;
auto res = engine->indexFactory().enhanceIndexDefinition( // normalize definition
input, normalized, create, collection->vocbase() // args
);
if (res.fail()) {
events::CreateIndex(collection->vocbase().name(), collection->name(), input,
res.errorNumber());
return res;
}
VPackSlice indexDef = normalized.slice();
if (ServerState::instance()->isCoordinator()) {
TRI_ASSERT(indexDef.isObject());
// check if there is an attempt to create a unique index on non-shard keys
if (create) {
Index::validateFields(indexDef);
auto v = indexDef.get(arangodb::StaticStrings::IndexUnique);
/* the following combinations of shardKeys and indexKeys are allowed/not
allowed:
shardKeys indexKeys
a a ok
a b not ok
a a b ok
a b a not ok
a b b not ok
a b a b ok
a b a b c ok
a b c a b not ok
a b c a b c ok
*/
if (v.isBoolean() && v.getBoolean()) {
// unique index, now check if fields and shard keys match
auto flds = indexDef.get(arangodb::StaticStrings::IndexFields);
if (flds.isArray() && collection->numberOfShards() > 1) {
std::vector<std::string> const& shardKeys = collection->shardKeys();
std::unordered_set<std::string> indexKeys;
size_t n = static_cast<size_t>(flds.length());
for (size_t i = 0; i < n; ++i) {
VPackSlice f = flds.at(i);
if (!f.isString()) {
// index attributes must be strings
events::CreateIndex(collection->vocbase().name(), collection->name(),
indexDef, TRI_ERROR_INTERNAL);
return Result(TRI_ERROR_INTERNAL,
"index field names should be strings");
}
indexKeys.emplace(f.copyString());
}
// all shard-keys must be covered by the index
for (auto& it : shardKeys) {
if (indexKeys.find(it) == indexKeys.end()) {
events::CreateIndex(collection->vocbase().name(), collection->name(),
indexDef, TRI_ERROR_CLUSTER_UNSUPPORTED);
return Result(TRI_ERROR_CLUSTER_UNSUPPORTED,
"shard key '" + it +
"' must be present in unique index");
}
}
}
}
}
}
TRI_ASSERT(!indexDef.isNone());
events::CreateIndex(collection->vocbase().name(), collection->name(),
indexDef, TRI_ERROR_NO_ERROR);
// ensure an index, coordinator case
if (ServerState::instance()->isCoordinator()) {
VPackBuilder tmp;
#ifdef USE_ENTERPRISE
Result res = Indexes::ensureIndexCoordinatorEE(collection, indexDef, create, tmp);
#else
Result res = Indexes::ensureIndexCoordinator(collection, indexDef, create, tmp);
#endif
if (!res.ok()) {
events::CreateIndex(collection->vocbase().name(), collection->name(),
indexDef, res.errorNumber());
return res;
} else if (tmp.slice().isNone()) {
// did not find a suitable index
int code = create ? TRI_ERROR_OUT_OF_MEMORY : TRI_ERROR_ARANGO_INDEX_NOT_FOUND;
events::CreateIndex(collection->vocbase().name(), collection->name(), indexDef, code);
return Result(code);
}
// flush estimates
collection->flushClusterIndexEstimates();
// the cluster won't set a proper id value
std::string iid = tmp.slice().get(StaticStrings::IndexId).copyString();
VPackBuilder b;
b.openObject();
b.add(StaticStrings::IndexId,
VPackValue(collection->name() + TRI_INDEX_HANDLE_SEPARATOR_CHR + iid));
b.close();
output = VPackCollection::merge(tmp.slice(), b.slice(), false);
events::CreateIndex(collection->vocbase().name(), collection->name(),
indexDef, res.errorNumber());
return res;
} else {
Result res = EnsureIndexLocal(collection, indexDef, create, output);
events::CreateIndex(collection->vocbase().name(), collection->name(),
indexDef, res.errorNumber());
return res;
}
}
arangodb::Result Indexes::createIndex(LogicalCollection* coll, Index::IndexType type,
std::vector<std::string> const& fields,
bool unique, bool sparse) {
VPackBuilder props;
props.openObject();
props.add(arangodb::StaticStrings::IndexType,
arangodb::velocypack::Value(Index::oldtypeName(type)));
props.add(arangodb::StaticStrings::IndexFields,
arangodb::velocypack::Value(VPackValueType::Array));
for (std::string const& field : fields) {
props.add(VPackValue(field));
}
props.close();
props.add(arangodb::StaticStrings::IndexUnique, arangodb::velocypack::Value(unique));
props.add(arangodb::StaticStrings::IndexSparse, arangodb::velocypack::Value(sparse));
props.close();
VPackBuilder ignored;
return ensureIndex(coll, props.slice(), true, ignored);
}
////////////////////////////////////////////////////////////////////////////////
/// @brief checks if argument is an index identifier
////////////////////////////////////////////////////////////////////////////////
static bool ExtractIndexHandle(VPackSlice const& arg,
std::string& collectionName, TRI_idx_iid_t& iid) {
TRI_ASSERT(collectionName.empty());
TRI_ASSERT(iid == 0);
if (arg.isNumber()) {
// numeric index id
iid = (TRI_idx_iid_t)arg.getUInt();
return true;
}
if (!arg.isString()) {
return false;
}
std::string str = arg.copyString();
size_t split;
if (arangodb::Index::validateHandle(str.data(), &split)) {
collectionName = std::string(str.data(), split);
iid = StringUtils::uint64(str.data() + split + 1, str.length() - split - 1);
return true;
}
if (arangodb::Index::validateId(str.data())) {
iid = StringUtils::uint64(str);
return true;
}
return false;
}
////////////////////////////////////////////////////////////////////////////////
/// @brief checks if argument is an index name
////////////////////////////////////////////////////////////////////////////////
static bool ExtractIndexName(VPackSlice const& arg, std::string& collectionName,
std::string& name) {
TRI_ASSERT(collectionName.empty());
TRI_ASSERT(name.empty());
if (!arg.isString()) {
return false;
}
std::string str = arg.copyString();
size_t split;
if (arangodb::Index::validateHandleName(str.data(), &split)) {
collectionName = std::string(str.data(), split);
name = std::string(str.data() + split + 1, str.length() - split - 1);
return true;
}
if (arangodb::Index::validateName(str.data())) {
name = str;
return true;
}
return false;
}
////////////////////////////////////////////////////////////////////////////////
/// @brief looks up an index identifier
////////////////////////////////////////////////////////////////////////////////
Result Indexes::extractHandle(arangodb::LogicalCollection const* collection,
arangodb::CollectionNameResolver const* resolver,
VPackSlice const& val, TRI_idx_iid_t& iid,
std::string& name) {
// reset the collection identifier
std::string collectionName;
// assume we are already loaded
TRI_ASSERT(collection != nullptr);
// extract the index identifier from a string
if (val.isString() || val.isNumber()) {
if (!ExtractIndexHandle(val, collectionName, iid) &&
!ExtractIndexName(val, collectionName, name)) {
return Result(TRI_ERROR_ARANGO_INDEX_HANDLE_BAD);
}
}
// extract the index identifier from an object
else if (val.isObject()) {
VPackSlice iidVal = val.get(StaticStrings::IndexId);
if (!ExtractIndexHandle(iidVal, collectionName, iid)) {
VPackSlice nameVal = val.get(StaticStrings::IndexName);
if (!ExtractIndexName(nameVal, collectionName, name)) {
return Result(TRI_ERROR_ARANGO_INDEX_HANDLE_BAD);
}
}
}
if (!collectionName.empty()) {
if (!EqualCollection(resolver, collectionName, collection)) {
// I wish this error provided me with more information!
// e.g. 'cannot access index outside the collection it was defined in'
return Result(TRI_ERROR_ARANGO_CROSS_COLLECTION_REQUEST);
}
}
return Result();
}
arangodb::Result Indexes::drop(LogicalCollection* collection, VPackSlice const& indexArg) {
TRI_ASSERT(collection);
if (ExecContext::CURRENT != nullptr) {
if (ExecContext::CURRENT->databaseAuthLevel() != auth::Level::RW ||
!ExecContext::CURRENT->canUseCollection(collection->name(), auth::Level::RW)) {
events::DropIndex(collection->vocbase().name(), collection->name(), "", TRI_ERROR_FORBIDDEN);
return TRI_ERROR_FORBIDDEN;
}
}
TRI_idx_iid_t iid = 0;
std::string name;
auto getHandle = [collection, &indexArg, &iid,
&name](CollectionNameResolver const* resolver,
transaction::Methods* trx = nullptr) -> Result {
Result res = Indexes::extractHandle(collection, resolver, indexArg, iid, name);
if (!res.ok()) {
events::DropIndex(collection->vocbase().name(), collection->name(), "",
res.errorNumber());
return res;
}
if (iid == 0 && !name.empty()) {
VPackBuilder builder;
res = methods::Indexes::getIndex(collection, indexArg, builder, trx);
if (!res.ok()) {
events::DropIndex(collection->vocbase().name(), collection->name(), "",
res.errorNumber());
return res;
}
VPackSlice idSlice = builder.slice().get(StaticStrings::IndexId);
Result res = Indexes::extractHandle(collection, resolver, idSlice, iid, name);
if (!res.ok()) {
events::DropIndex(collection->vocbase().name(), collection->name(), "",
res.errorNumber());
}
}
return res;
};
if (ServerState::instance()->isCoordinator()) {
CollectionNameResolver resolver(collection->vocbase());
Result res = getHandle(&resolver);
if (!res.ok()) {
return res;
}
// flush estimates
collection->flushClusterIndexEstimates();
#ifdef USE_ENTERPRISE
res = Indexes::dropCoordinatorEE(collection, iid);
#else
res = ClusterInfo::instance()->dropIndexCoordinator( // drop index
collection->vocbase().name(), std::to_string(collection->id()), iid, 0.0 // args
);
#endif
events::DropIndex(collection->vocbase().name(), collection->name(),
std::to_string(iid), res.errorNumber());
return res;
} else {
READ_LOCKER(readLocker, collection->vocbase()._inventoryLock);
SingleCollectionTransaction trx(
transaction::V8Context::CreateWhenRequired(collection->vocbase(), false),
*collection, AccessMode::Type::EXCLUSIVE);
Result res = trx.begin();
if (!res.ok()) {
events::DropIndex(collection->vocbase().name(), collection->name(), "",
res.errorNumber());
return res;
}
LogicalCollection* col = trx.documentCollection();
res = getHandle(trx.resolver(), &trx);
if (!res.ok()) {
return res;
}
std::shared_ptr<Index> idx = collection->lookupIndex(iid);
if (!idx || idx->id() == 0) {
events::DropIndex(collection->vocbase().name(), collection->name(),
std::to_string(iid), TRI_ERROR_ARANGO_INDEX_NOT_FOUND);
return Result(TRI_ERROR_ARANGO_INDEX_NOT_FOUND);
}
if (!idx->canBeDropped()) {
events::DropIndex(collection->vocbase().name(), collection->name(),
std::to_string(iid), TRI_ERROR_FORBIDDEN);
return Result(TRI_ERROR_FORBIDDEN);
}
bool ok = col->dropIndex(idx->id());
int code = ok ? TRI_ERROR_NO_ERROR : TRI_ERROR_FAILED;
events::DropIndex(collection->vocbase().name(), collection->name(),
std::to_string(iid), code);
return Result(code);
}
}
Reference in New Issue View Git Blame Copy Permalink