Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Resolves issue #11: Now supports server version 3.2. Changes made in this PR: #195

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Resolves issue #11: Now supports server version 3.2. Changes made in this PR: #195

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d2f4abc
Now supports server version 3.2. Changes made in this PR:
May 31, 2019
79d37c4
Fix the linux build failure
Jun 10, 2019
aacb386
Piggy back this PR to do some misc error tracing improvements
Jun 17, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions src/Constants.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,26 @@ const char* METADATA_VERSION_FIELD = "metadata version";
const char* NS_FIELD = "ns";
const char* QUERY_FIELD = "query";
const char* CURRENTLY_PROCESSING_DOC_FIELD = "currently processing document";
const char* GET_MORE_CMD_BATCH_SIZE_FIELD = "batchSize";
const char* GET_MORE_CMD_CURSOR_ID_FIELD = "getMore";
const char* GET_MORE_CMD_CURSOR_COLLECTION_FIELD = "collection";
const char* FIND_CMD_FIND_FIELD = "find";
const char* FIND_CMD_FILTER_FIELD = "filter";
const char* FIND_CMD_SORT_FIELD = "sort";
const char* FIND_CMD_PROJECTION_FIELD = "projection";
const char* FIND_CMD_HINT_FIELD = "hint";
const char* FIND_CMD_SKIP_FIELD = "skip";
const char* FIND_CMD_LIMIT_FIELD = "limit";
const char* FIND_CMD_BATCH_SIZE_FIELD = "batchSize";
const char* FIND_CMD_MAX_TIME_MS_FIELD = "maxTimeMS";
const char* FIND_CMD_REPLY_CURSOR_FIELD = "cursor";
const char* FIND_CMD_REPLY_CURSOR_ID_FIELD = "id";
const char* FIND_CMD_REPLY_CURSOR_NS_FIELD = "ns";
const char* FIND_CMD_REPLY_CURSOR_FIRST_BATCH_FIELD = "firstBatch";
const char* FIND_CMD_REPLY_CURSOR_NEXT_BATCH_FIELD = "nextBatch";
const char* KILL_CURSORS_CMD_KILL_CURSORS_FIELD = "killCursors";
const char* KILL_CURSORS_CMD_CURSORS_FIELD = "cursors";
const char* EXPLAIN_CMD_FIELD = "explain";

const std::string RENAME = "$rename";
const std::string SET = "$set";
Expand Down
20 changes: 20 additions & 0 deletions src/Constants.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,26 @@ extern const char* METADATA_VERSION_FIELD;
extern const char* NS_FIELD;
extern const char* QUERY_FIELD;
extern const char* CURRENTLY_PROCESSING_DOC_FIELD;
extern const char* GET_MORE_CMD_BATCH_SIZE_FIELD;
extern const char* GET_MORE_CMD_CURSOR_ID_FIELD;
extern const char* GET_MORE_CMD_CURSOR_COLLECTION_FIELD;
extern const char* FIND_CMD_FIND_FIELD;
extern const char* FIND_CMD_FILTER_FIELD;
extern const char* FIND_CMD_SORT_FIELD;
extern const char* FIND_CMD_PROJECTION_FIELD;
extern const char* FIND_CMD_HINT_FIELD;
extern const char* FIND_CMD_SKIP_FIELD;
extern const char* FIND_CMD_LIMIT_FIELD;
extern const char* FIND_CMD_BATCH_SIZE_FIELD;
extern const char* FIND_CMD_MAX_TIME_MS_FIELD;
extern const char* FIND_CMD_REPLY_CURSOR_FIELD;
extern const char* FIND_CMD_REPLY_CURSOR_ID_FIELD;
extern const char* FIND_CMD_REPLY_CURSOR_NS_FIELD;
extern const char* FIND_CMD_REPLY_CURSOR_FIRST_BATCH_FIELD;
extern const char* FIND_CMD_REPLY_CURSOR_NEXT_BATCH_FIELD;
extern const char* KILL_CURSORS_CMD_KILL_CURSORS_FIELD;
extern const char* KILL_CURSORS_CMD_CURSORS_FIELD;
extern const char* EXPLAIN_CMD_FIELD;

// Mongo Operators
extern const std::string RENAME;
Expand Down
5 changes: 3 additions & 2 deletions src/Cursor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,11 +31,12 @@ struct Cursor : ReferenceCounted<Cursor>, NonCopyable {
Reference<PlanCheckpoint> checkpoint;
int64_t id;
int32_t returned;
int32_t limit;
std::map<int64_t, Reference<Cursor>>* siblings;
time_t expiry;

Cursor(FutureStream<Reference<ScanReturnedContext>> docs, Reference<PlanCheckpoint> checkpoint)
: docs(docs), checkpoint(checkpoint), returned(0) {
Cursor(FutureStream<Reference<ScanReturnedContext>> docs, Reference<PlanCheckpoint> checkpoint, int32_t limit = 0)
: docs(docs), checkpoint(checkpoint), returned(0), limit(limit) {
id = g_random->randomInt64(INT64_MIN, INT64_MAX);
expiry = time(nullptr) + DOCLAYER_KNOBS->CURSOR_EXPIRY;
siblings = nullptr;
Expand Down
6 changes: 3 additions & 3 deletions src/Ext.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,9 @@
#include "bson.h"
#include "flow/flow.h"

#define EXT_SERVER_VERSION "3.0.0"
#define EXT_SERVER_VERSION_ARRAY BSON_ARRAY(3 << 0 << 0)
#define EXT_SERVER_VERSION "3.2.0"
#define EXT_SERVER_VERSION_ARRAY BSON_ARRAY(3 << 2 << 0)
#define EXT_MIN_WIRE_VERSION 0
#define EXT_MAX_WIRE_VERSION 3
#define EXT_MAX_WIRE_VERSION 4

#endif /* _EXT_H_ */
240 changes: 240 additions & 0 deletions src/ExtCmd.actor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -919,6 +919,246 @@ struct AvailableQueryOptionsCmd {
};
REGISTER_CMD(AvailableQueryOptionsCmd, "availablequeryoptions");

ACTOR static Future<Reference<ExtMsgReply>> doGetMore(Reference<ExtMsgReply> reply,
Reference<ExtConnection> ec,
Namespace ns,
int64_t cursorID,
int32_t batchSize = 101) {
state Reference<Cursor> cursor = ec->cursors[cursorID];

if (cursor) {
try {
int32_t returned =
wait(addDocumentsFromCursor(cursor, reply, ns, batchSize, AddDocsFromCursorCaller::GET_MORE_CMD));
reply->replyHeader.startingFrom = cursor->returned - returned;
reply->addResponseFlag(8 /*0b1000*/);
cursor->refresh();
} catch (Error& e) {
reply->setError(e);
}
} else {
reply->addDocument(BSON("ok" << 0 << "errmsg"
<< "CursorNotFound"
<< "code" << 43));
reply->addResponseFlag(0 /*0b0000*/); // set flag to 0 indicating cursor does not exist
}
return reply;
}

struct GetMoreCmd {
static const char* name;
static Future<Reference<ExtMsgReply>> call(Reference<ExtConnection> nmc,
Reference<ExtMsgQuery> query,
Reference<ExtMsgReply> reply) {
if (!query->query.getField(DocLayerConstants::GET_MORE_CMD_CURSOR_ID_FIELD).isNumber() ||
!query->query.getField(DocLayerConstants::GET_MORE_CMD_CURSOR_COLLECTION_FIELD).isString()) {
TraceEvent(SevWarn, "WireBadGetMoreCmd").detail("query", query->query.toString()).suppressFor(1.0);
throw wire_protocol_mismatch();
}

Namespace ns = std::make_pair(
query->ns.first, query->query.getField(DocLayerConstants::GET_MORE_CMD_CURSOR_COLLECTION_FIELD).str());

bson::BSONElement batchSizeField = query->query.getField(DocLayerConstants::GET_MORE_CMD_BATCH_SIZE_FIELD);
int64_t cursorID = query->query.getField(DocLayerConstants::GET_MORE_CMD_CURSOR_ID_FIELD).numberLong();
return doGetMore(reply, nmc, query->ns, cursorID, batchSizeField.isNumber() ? batchSizeField.numberInt() : 101);
}
};

REGISTER_CMD(GetMoreCmd, "getmore");

ACTOR static Future<Reference<ExtMsgReply>> doExplain(
Reference<ExtConnection> ec,
Reference<ExtMsgReply> reply,
Namespace ns,
bson::BSONObj filter,
bson::BSONObj projection,
Optional<bson::BSONObj> ordering = Optional<bson::BSONObj>(),
int32_t limit = 0, // hard limit on how many a cursor can return before got closed
int32_t skip = 0,
int32_t batchSize = 101, // how many docs to return in first batch
Optional<int64_t> maxTimeMS = Optional<int64_t>(), // TODO make use of maxTimeMS
Optional<bson::BSONObj> hint = Optional<bson::BSONObj>() // TODO make use of hint
) {
try {
state Reference<DocTransaction> dtr = ec->getOperationTransaction();
state Reference<UnboundCollectionContext> cx = wait(ec->mm->getUnboundCollectionContext(dtr, ns, true));
state Reference<Cursor> cursor;

state Reference<Plan> plan = planQuery(cx, filter);
if (!ordering.present() && skip > 0)
plan = ref(new SkipPlan(skip, plan));
plan = planProjection(plan, projection, ordering);
plan = ec->wrapOperationPlan(plan, true, cx);
if (ordering.present()) {
plan = ref(new SortPlan(plan, ordering.get()));
if (skip > 0)
plan = ref(new SkipPlan(skip, plan));
}
reply->addDocument(BSON("ok" << 1 << "explanation" << plan->describe()));
} catch (Error& e) {
if (e.code() != error_code_end_of_stream) {
reply->setError(e.what(), e.code());
}
}
return reply;
}

struct ExplainCmd {
static const char* name;
static Future<Reference<ExtMsgReply>> call(Reference<ExtConnection> nmc,
Reference<ExtMsgQuery> msg,
Reference<ExtMsgReply> reply) {
bson::BSONElement explainField = msg->query.getField(DocLayerConstants::EXPLAIN_CMD_FIELD);
if (!explainField.isABSONObj() ||
!explainField.embeddedObject().getField(DocLayerConstants::FIND_CMD_FIND_FIELD).isString()) {
throw wire_protocol_mismatch();
}

bson::BSONObj explainObj = explainField.embeddedObject();

Namespace ns = std::make_pair(msg->ns.first, explainObj.getStringField(DocLayerConstants::FIND_CMD_FIND_FIELD));

bson::BSONElement filterField = explainObj.getField(DocLayerConstants::FIND_CMD_FILTER_FIELD);
bson::BSONElement projectionField = explainObj.getField(DocLayerConstants::FIND_CMD_PROJECTION_FIELD);
bson::BSONElement orderingField = explainObj.getField(DocLayerConstants::FIND_CMD_SORT_FIELD);
bson::BSONElement limitField = explainObj.getField(DocLayerConstants::FIND_CMD_LIMIT_FIELD);
bson::BSONElement skipField = explainObj.getField(DocLayerConstants::FIND_CMD_SKIP_FIELD);
bson::BSONElement batchSizeField = explainObj.getField(DocLayerConstants::FIND_CMD_BATCH_SIZE_FIELD);
bson::BSONElement maxTimeMSField = explainObj.getField(DocLayerConstants::FIND_CMD_MAX_TIME_MS_FIELD);
bson::BSONElement hintField = explainObj.getField(DocLayerConstants::FIND_CMD_HINT_FIELD);

return doExplain(
nmc, reply, ns, filterField.isABSONObj() ? filterField.embeddedObject() : bson::BSONObj(),
projectionField.isABSONObj() ? projectionField.embeddedObject() : bson::BSONObj(),
orderingField.isABSONObj() ? Optional<bson::BSONObj>(orderingField.embeddedObject())
: Optional<bson::BSONObj>(),
limitField.isNumber() ? limitField.numberInt() : 0, skipField.isNumber() ? skipField.numberInt() : 0,
batchSizeField.isNumber() ? batchSizeField.numberInt() : 101,
maxTimeMSField.isNumber() ? Optional<int64_t>(maxTimeMSField.numberLong()) : Optional<int64_t>(),
hintField.isABSONObj() ? Optional<bson::BSONObj>(hintField.embeddedObject()) : Optional<bson::BSONObj>());
}
};

REGISTER_CMD(ExplainCmd, "explain");

ACTOR static Future<Reference<ExtMsgReply>> doFind(
Reference<ExtConnection> ec,
Reference<ExtMsgReply> reply,
Namespace ns,
bson::BSONObj filter,
bson::BSONObj projection,
Optional<bson::BSONObj> ordering = Optional<bson::BSONObj>(),
int32_t limit = 0, // hard limit on how many a cursor can return before got closed
int32_t skip = 0,
int32_t batchSize = 101, // how many docs to return in first batch
Optional<int64_t> maxTimeMS = Optional<int64_t>(), // TODO make use of maxTimeMS
Optional<bson::BSONObj> hint = Optional<bson::BSONObj>() // TODO make use of hint
) {
try {
state Reference<DocTransaction> dtr = ec->getOperationTransaction();
state Reference<UnboundCollectionContext> cx = wait(ec->mm->getUnboundCollectionContext(dtr, ns, true));
state Reference<Cursor> cursor;

state Reference<Plan> plan = planQuery(cx, filter);
if (!ordering.present() && skip > 0)
plan = ref(new SkipPlan(skip, plan));
plan = planProjection(plan, projection, ordering);
plan = ec->wrapOperationPlan(plan, true, cx);
if (ordering.present()) {
plan = ref(new SortPlan(plan, ordering.get()));
if (skip > 0)
plan = ref(new SkipPlan(skip, plan));
}
// TODO remove the $explain operator handling here since it should have been deprecated in this command from 3.2
if (filter.hasField("$explain")) {
reply->addDocument(BSON("explanation" << plan->describe()));
return reply;
}

Reference<PlanCheckpoint> outerCheckpoint(new PlanCheckpoint);

// Add a new cursor to the server's cursor collection
cursor = Cursor::add(ec->cursors, Reference<Cursor>(new Cursor(plan->execute(outerCheckpoint.getPtr(), dtr),
outerCheckpoint, limit)));
int32_t returned =
wait(addDocumentsFromCursor(cursor, reply, ns, batchSize, AddDocsFromCursorCaller::FIND_CMD));
} catch (Error& e) {
if (e.code() != error_code_end_of_stream) {
reply->setError(e.what(), e.code());
}
}
return reply;
}

struct FindCmd {
static const char* name;
static Future<Reference<ExtMsgReply>> call(Reference<ExtConnection> ec,
Reference<ExtMsgQuery> msg,
Reference<ExtMsgReply> reply) {
bson::BSONElement findField = msg->query.getField(DocLayerConstants::FIND_CMD_FIND_FIELD);
if (!findField.isString()) {
throw wire_protocol_mismatch();
}

Namespace ns = std::make_pair(msg->ns.first, findField.str());

bson::BSONElement filterField = msg->query.getField(DocLayerConstants::FIND_CMD_FILTER_FIELD);
bson::BSONElement projectionField = msg->query.getField(DocLayerConstants::FIND_CMD_PROJECTION_FIELD);
bson::BSONElement orderingField = msg->query.getField(DocLayerConstants::FIND_CMD_SORT_FIELD);
bson::BSONElement limitField = msg->query.getField(DocLayerConstants::FIND_CMD_LIMIT_FIELD);
bson::BSONElement skipField = msg->query.getField(DocLayerConstants::FIND_CMD_SKIP_FIELD);
bson::BSONElement batchSizeField = msg->query.getField(DocLayerConstants::FIND_CMD_BATCH_SIZE_FIELD);
bson::BSONElement maxTimeMSField = msg->query.getField(DocLayerConstants::FIND_CMD_MAX_TIME_MS_FIELD);
bson::BSONElement hintField = msg->query.getField(DocLayerConstants::FIND_CMD_HINT_FIELD);

return doFind(
ec, reply, ns, filterField.isABSONObj() ? filterField.embeddedObject() : bson::BSONObj(),
projectionField.isABSONObj() ? projectionField.embeddedObject() : bson::BSONObj(),
orderingField.isABSONObj() ? Optional<bson::BSONObj>(orderingField.embeddedObject())
: Optional<bson::BSONObj>(),
limitField.isNumber() ? limitField.numberInt() : 0, skipField.isNumber() ? skipField.numberInt() : 0,
batchSizeField.isNumber() ? batchSizeField.numberInt() : 101,
maxTimeMSField.isNumber() ? Optional<int64_t>(maxTimeMSField.numberLong()) : Optional<int64_t>(),
hintField.isABSONObj() ? Optional<bson::BSONObj>(hintField.embeddedObject()) : Optional<bson::BSONObj>());
}
};

REGISTER_CMD(FindCmd, "find");

static Future<Reference<ExtMsgReply>> doKillCursors(Reference<ExtConnection> ec,
Reference<ExtMsgReply> reply,
bson::BSONArray cursorIds) {
for (bson::BSONObj::iterator i = cursorIds.begin(); i.more();) {
bson::BSONElement cursorId = i.next();
if (!cursorId.isNumber()) {
throw wire_protocol_mismatch();
}
Cursor::pluck(ec->cursors[cursorId.numberLong()]);
}
reply->addDocument(BSON("ok" << 1));
return reply;
}

struct KillCursorsCmd {
static const char* name;
static Future<Reference<ExtMsgReply>> call(Reference<ExtConnection> ec,
Reference<ExtMsgQuery> msg,
Reference<ExtMsgReply> reply) {
bson::BSONElement killCursors = msg->query.getField(DocLayerConstants::KILL_CURSORS_CMD_KILL_CURSORS_FIELD);
if (!killCursors.isString()) {
throw wire_protocol_mismatch();
}
bson::BSONElement cursorsField = msg->query.getField(DocLayerConstants::KILL_CURSORS_CMD_CURSORS_FIELD);
if (cursorsField.type() != bson::BSONType::Array) {
throw wire_protocol_mismatch();
}
return doKillCursors(ec, reply, bson::BSONArray(cursorsField.Obj()));
}
};

REGISTER_CMD(KillCursorsCmd, "killcursors");

struct GetMemoryUsageCmd {
static const char* name;
static Future<Reference<ExtMsgReply>> call(Reference<ExtConnection> nmc,
Expand Down
Loading