mirror of
https://github.com/dragonflydb/dragonfly.git
synced 2025-05-11 10:25:47 +02:00
chore: Del and NUMINCRBY use json::Path (#2655)
* chore: Del and NUMINCRBY use json::Path Also, fix various protocol bugs when we sent simple string instead of sending bulk strings. Fixed a typo in path.cc that lead to a data race bug. Finally, flip the flag in regression tests to start covering json::Path code and added test coverage for the data race bug --------- Signed-off-by: Roman Gershman <roman@dragonflydb.io>
This commit is contained in:
Roman Gershman
GitHub
parent
d54f2201ae
commit
91c299b33e
5 changed files with 201 additions and 152 deletions
|
|
@ -365,8 +365,9 @@ auto Dfs::MutateStep(const PathSegment& segment, const MutateCallback& cb, JsonT
|
|||
if (segment.index() >= node->size()) {
|
||||
return make_unexpected(OUT_OF_BOUNDS);
|
||||
}
|
||||
if (Mutate(cb, nullopt, &node[segment.index()])) {
|
||||
node->erase(node->array_range().begin() + segment.index());
|
||||
auto it = node->array_range().begin() + segment.index();
|
||||
if (Mutate(cb, nullopt, &*it)) {
|
||||
node->erase(it);
|
||||
}
|
||||
} break;
|
||||
|
||||
|
|
@ -475,10 +476,12 @@ nonstd::expected<json::Path, string> ParsePath(string_view path) {
|
|||
return driver.TakePath();
|
||||
}
|
||||
|
||||
void MutatePath(const Path& path, MutateCallback callback, JsonType* json) {
|
||||
unsigned MutatePath(const Path& path, MutateCallback callback, JsonType* json) {
|
||||
if (path.empty())
|
||||
return;
|
||||
Dfs().Mutate(path, callback, json);
|
||||
return 0;
|
||||
Dfs dfs;
|
||||
dfs.Mutate(path, callback, json);
|
||||
return dfs.matches();
|
||||
}
|
||||
|
||||
} // namespace dfly::json
|
||||
|
|
|
|||
|
|
@ -95,7 +95,9 @@ using PathCallback = absl::FunctionRef<void(std::optional<std::string_view>, con
|
|||
using MutateCallback = absl::FunctionRef<bool(std::optional<std::string_view>, JsonType*)>;
|
||||
|
||||
void EvaluatePath(const Path& path, const JsonType& json, PathCallback callback);
|
||||
void MutatePath(const Path& path, MutateCallback callback, JsonType* json);
|
||||
|
||||
// returns number of matches found with the given path.
|
||||
unsigned MutatePath(const Path& path, MutateCallback callback, JsonType* json);
|
||||
nonstd::expected<Path, std::string> ParsePath(std::string_view path);
|
||||
|
||||
} // namespace dfly::json
|
||||
|
|
|
|||
|
|
@ -43,8 +43,7 @@ using OptBool = optional<bool>;
|
|||
using OptLong = optional<long>;
|
||||
using OptSizeT = optional<size_t>;
|
||||
using OptString = optional<string>;
|
||||
using JsonReplaceCb = function<void(const JsonExpression::path_node_type&, JsonType&)>;
|
||||
using JsonReplaceVerify = std::function<OpStatus(JsonType&)>;
|
||||
using JsonReplaceVerify = std::function<void(JsonType&)>;
|
||||
using CI = CommandId;
|
||||
|
||||
static const char DefaultJsonPath[] = "$";
|
||||
|
|
@ -75,10 +74,6 @@ inline JsonType Evaluate(const json::Path& expr, const JsonType& obj) {
|
|||
return res;
|
||||
}
|
||||
|
||||
inline OpStatus JsonReplaceVerifyNoOp(JsonType&) {
|
||||
return OpStatus::OK;
|
||||
}
|
||||
|
||||
facade::OpStatus SetJson(const OpArgs& op_args, string_view key, JsonType&& value) {
|
||||
auto& db_slice = op_args.shard->db_slice();
|
||||
|
||||
|
|
@ -156,7 +151,7 @@ void PrintOptVec(ConnectionContext* cntx, const OpResult<vector<optional<T>>>& r
|
|||
}
|
||||
}
|
||||
|
||||
error_code JsonReplace(JsonType& instance, string_view path, JsonReplaceCb callback) {
|
||||
error_code JsonReplace(JsonType& instance, string_view path, json::MutateCallback callback) {
|
||||
using evaluator_t = jsonpath::detail::jsonpath_evaluator<JsonType, JsonType&>;
|
||||
using value_type = evaluator_t::value_type;
|
||||
using reference = evaluator_t::reference;
|
||||
|
|
@ -173,8 +168,9 @@ error_code JsonReplace(JsonType& instance, string_view path, JsonReplaceCb callb
|
|||
}
|
||||
|
||||
jsonpath::detail::dynamic_resources<value_type, reference> resources;
|
||||
auto f = [&callback](const json_selector_t::path_node_type& path, reference val) {
|
||||
callback(path, val);
|
||||
|
||||
auto f = [&callback](const jsonpath::basic_path_node<char>& path, JsonType& val) {
|
||||
callback(jsonpath::to_string(path), &val);
|
||||
};
|
||||
|
||||
expr.evaluate(resources, instance, json_selector_t::path_node_type{}, instance, f,
|
||||
|
|
@ -184,7 +180,7 @@ error_code JsonReplace(JsonType& instance, string_view path, JsonReplaceCb callb
|
|||
|
||||
// jsoncons version
|
||||
OpStatus UpdateEntry(const OpArgs& op_args, std::string_view key, std::string_view path,
|
||||
JsonReplaceCb callback, JsonReplaceVerify verify_op = JsonReplaceVerifyNoOp) {
|
||||
json::MutateCallback callback, JsonReplaceVerify verify_op = {}) {
|
||||
auto it_res = op_args.shard->db_slice().FindMutable(op_args.db_cntx, key, OBJ_JSON);
|
||||
if (!it_res.ok()) {
|
||||
return it_res.status();
|
||||
|
|
@ -206,13 +202,14 @@ OpStatus UpdateEntry(const OpArgs& op_args, std::string_view key, std::string_vi
|
|||
}
|
||||
|
||||
// Make sure that we don't have other internal issue with the operation
|
||||
OpStatus res = verify_op(json_entry);
|
||||
if (res == OpStatus::OK) {
|
||||
it_res->post_updater.Run();
|
||||
op_args.shard->search_indices()->AddDoc(key, op_args.db_cntx, entry_it->second);
|
||||
if (verify_op) {
|
||||
verify_op(json_entry);
|
||||
}
|
||||
|
||||
return res;
|
||||
it_res->post_updater.Run();
|
||||
op_args.shard->search_indices()->AddDoc(key, op_args.db_cntx, entry_it->second);
|
||||
|
||||
return OpStatus::OK;
|
||||
}
|
||||
|
||||
// json::Path version.
|
||||
|
|
@ -428,13 +425,13 @@ void SendJsonValue(RedisReplyBuilder* rb, const JsonType& j) {
|
|||
} else if (j.is_null()) {
|
||||
rb->SendNull();
|
||||
} else if (j.is_string()) {
|
||||
rb->SendSimpleString(j.as_string_view());
|
||||
rb->SendBulkString(j.as_string_view());
|
||||
} else if (j.is_object()) {
|
||||
rb->StartArray(j.size() + 1);
|
||||
rb->SendSimpleString("{");
|
||||
for (const auto& item : j.object_range()) {
|
||||
rb->StartArray(2);
|
||||
rb->SendSimpleString(item.key());
|
||||
rb->SendBulkString(item.key());
|
||||
SendJsonValue(rb, item.value());
|
||||
}
|
||||
} else if (j.is_array()) {
|
||||
|
|
@ -587,30 +584,21 @@ OpResult<vector<OptBool>> OpToggle(const OpArgs& op_args, string_view key, strin
|
|||
JsonPathV2 expression) {
|
||||
vector<OptBool> vec;
|
||||
OpStatus status;
|
||||
if (std::holds_alternative<json::Path>(expression)) {
|
||||
auto cb = [&vec](optional<string_view>, JsonType* val) {
|
||||
if (val->is_bool()) {
|
||||
bool next_val = val->as_bool() ^ true;
|
||||
*val = next_val;
|
||||
vec.emplace_back(next_val);
|
||||
} else {
|
||||
vec.emplace_back(nullopt);
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
if (holds_alternative<json::Path>(expression)) {
|
||||
const json::Path& expr = std::get<json::Path>(expression);
|
||||
auto cb = [&vec](optional<string_view>, JsonType* val) {
|
||||
if (val->is_bool()) {
|
||||
bool next_val = val->as_bool() ^ true;
|
||||
*val = next_val;
|
||||
vec.emplace_back(next_val);
|
||||
} else {
|
||||
vec.emplace_back(nullopt);
|
||||
}
|
||||
return false;
|
||||
};
|
||||
status = UpdateEntry(op_args, key, expr, cb);
|
||||
} else {
|
||||
auto cb = [&vec](const auto&, JsonType& val) {
|
||||
if (val.is_bool()) {
|
||||
bool current_val = val.as_bool() ^ true;
|
||||
val = current_val;
|
||||
vec.emplace_back(current_val);
|
||||
} else {
|
||||
vec.emplace_back(nullopt);
|
||||
}
|
||||
};
|
||||
|
||||
status = UpdateEntry(op_args, key, path, cb);
|
||||
}
|
||||
if (status != OpStatus::OK) {
|
||||
|
|
@ -619,41 +607,65 @@ OpResult<vector<OptBool>> OpToggle(const OpArgs& op_args, string_view key, strin
|
|||
return vec;
|
||||
}
|
||||
|
||||
template <typename Op>
|
||||
enum ArithmeticOpType { OP_ADD, OP_MULTIPLY };
|
||||
|
||||
void BinOpApply(double num, bool num_is_double, ArithmeticOpType op, JsonType* val,
|
||||
bool* overflow) {
|
||||
double result = 0;
|
||||
switch (op) {
|
||||
case OP_ADD:
|
||||
result = val->as<double>() + num;
|
||||
break;
|
||||
case OP_MULTIPLY:
|
||||
result = val->as<double>() * num;
|
||||
break;
|
||||
}
|
||||
|
||||
if (isinf(result)) {
|
||||
*overflow = true;
|
||||
return;
|
||||
}
|
||||
|
||||
if (val->is_double() || num_is_double) {
|
||||
*val = result;
|
||||
} else {
|
||||
*val = (uint64_t)result;
|
||||
}
|
||||
*overflow = false;
|
||||
}
|
||||
|
||||
OpResult<string> OpDoubleArithmetic(const OpArgs& op_args, string_view key, string_view path,
|
||||
double num, Op arithmetic_op) {
|
||||
double num, ArithmeticOpType op_type, JsonPathV2 expression) {
|
||||
bool is_result_overflow = false;
|
||||
double int_part;
|
||||
bool has_fractional_part = (modf(num, &int_part) != 0);
|
||||
JsonType output(json_array_arg);
|
||||
OpStatus status;
|
||||
|
||||
auto cb = [&](const auto&, JsonType& val) {
|
||||
if (val.is_number()) {
|
||||
double result = arithmetic_op(val.as<double>(), num);
|
||||
if (isinf(result)) {
|
||||
auto cb = [&](optional<string_view>, JsonType* val) {
|
||||
if (val->is_number()) {
|
||||
bool res = false;
|
||||
BinOpApply(num, has_fractional_part, op_type, val, &res);
|
||||
if (res) {
|
||||
is_result_overflow = true;
|
||||
return;
|
||||
}
|
||||
|
||||
if (val.is_double() || has_fractional_part) {
|
||||
val = result;
|
||||
} else {
|
||||
val = (uint64_t)result;
|
||||
output.push_back(*val);
|
||||
}
|
||||
output.push_back(val);
|
||||
} else {
|
||||
output.push_back(JsonType::null());
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
auto verifier = [&is_result_overflow](JsonType&) {
|
||||
if (is_result_overflow) {
|
||||
return OpStatus::INVALID_NUMERIC_RESULT;
|
||||
}
|
||||
return OpStatus::OK;
|
||||
};
|
||||
if (holds_alternative<json::Path>(expression)) {
|
||||
const json::Path& path = std::get<json::Path>(expression);
|
||||
status = UpdateEntry(op_args, key, path, std::move(cb));
|
||||
} else {
|
||||
status = UpdateEntry(op_args, key, path, std::move(cb));
|
||||
}
|
||||
|
||||
OpStatus status = UpdateEntry(op_args, key, path, cb, verifier);
|
||||
if (is_result_overflow)
|
||||
return OpStatus::INVALID_NUMERIC_RESULT;
|
||||
if (status != OpStatus::OK) {
|
||||
return status;
|
||||
}
|
||||
|
|
@ -661,36 +673,46 @@ OpResult<string> OpDoubleArithmetic(const OpArgs& op_args, string_view key, stri
|
|||
return output.as_string();
|
||||
}
|
||||
|
||||
OpResult<long> OpDel(const OpArgs& op_args, string_view key, string_view path) {
|
||||
long total_deletions = 0;
|
||||
if (path.empty()) {
|
||||
// If expression is nullopt, then the whole key should be deleted, otherwise deletes
|
||||
// items specified by the expression/path.
|
||||
OpResult<long> OpDel(const OpArgs& op_args, string_view key, string_view path,
|
||||
optional<JsonPathV2> expression) {
|
||||
if (!expression || path.empty()) {
|
||||
auto& db_slice = op_args.shard->db_slice();
|
||||
auto it = db_slice.FindMutable(op_args.db_cntx, key).it; // post_updater will run immediately
|
||||
total_deletions += long(db_slice.Del(op_args.db_cntx.db_index, it));
|
||||
return total_deletions;
|
||||
return long(db_slice.Del(op_args.db_cntx.db_index, it));
|
||||
}
|
||||
|
||||
OpResult<JsonType*> result = GetJson(op_args, key);
|
||||
if (!result) {
|
||||
return total_deletions;
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (holds_alternative<json::Path>(*expression)) {
|
||||
const json::Path& path = get<json::Path>(*expression);
|
||||
long deletions = json::MutatePath(
|
||||
path, [](optional<string_view>, JsonType* val) { return true; }, *result);
|
||||
return deletions;
|
||||
}
|
||||
|
||||
vector<string> deletion_items;
|
||||
auto cb = [&](const JsonExpression::path_node_type& path, JsonType& val) {
|
||||
deletion_items.emplace_back(jsonpath::to_string(path));
|
||||
auto cb = [&](const auto& path, JsonType* val) {
|
||||
deletion_items.emplace_back(*path);
|
||||
return false;
|
||||
};
|
||||
|
||||
JsonType& json_entry = *(result.value());
|
||||
error_code ec = JsonReplace(json_entry, path, cb);
|
||||
error_code ec = JsonReplace(json_entry, path, std::move(cb));
|
||||
if (ec) {
|
||||
VLOG(1) << "Failed to evaluate expression on json with error: " << ec.message();
|
||||
return total_deletions;
|
||||
return 0;
|
||||
}
|
||||
|
||||
if (deletion_items.empty()) {
|
||||
return total_deletions;
|
||||
return 0;
|
||||
}
|
||||
|
||||
long total_deletions = 0;
|
||||
JsonType patch(json_array_arg, {});
|
||||
reverse(deletion_items.begin(), deletion_items.end()); // deletion should finish at root keys.
|
||||
for (const auto& item : deletion_items) {
|
||||
|
|
@ -744,18 +766,19 @@ OpResult<vector<StringVec>> OpObjKeys(const OpArgs& op_args, string_view key,
|
|||
OpResult<vector<OptSizeT>> OpStrAppend(const OpArgs& op_args, string_view key, string_view path,
|
||||
const vector<string_view>& strs) {
|
||||
vector<OptSizeT> vec;
|
||||
auto cb = [&](const auto&, JsonType& val) {
|
||||
if (val.is_string()) {
|
||||
string new_val = val.as_string();
|
||||
auto cb = [&](const auto&, JsonType* val) {
|
||||
if (val->is_string()) {
|
||||
string new_val = val->as_string();
|
||||
for (auto& str : strs) {
|
||||
new_val += str;
|
||||
}
|
||||
|
||||
val = new_val;
|
||||
*val = new_val;
|
||||
vec.emplace_back(new_val.size());
|
||||
} else {
|
||||
vec.emplace_back(nullopt);
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
OpStatus status = UpdateEntry(op_args, key, path, cb);
|
||||
|
|
@ -770,20 +793,21 @@ OpResult<vector<OptSizeT>> OpStrAppend(const OpArgs& op_args, string_view key, s
|
|||
// Clears containers(arrays or objects) and zeroing numbers.
|
||||
OpResult<long> OpClear(const OpArgs& op_args, string_view key, string_view path) {
|
||||
long clear_items = 0;
|
||||
auto cb = [&clear_items](const auto& path, JsonType& val) {
|
||||
if (!(val.is_object() || val.is_array() || val.is_number())) {
|
||||
return;
|
||||
auto cb = [&clear_items](const auto& path, JsonType* val) {
|
||||
if (!(val->is_object() || val->is_array() || val->is_number())) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (val.is_object()) {
|
||||
val.erase(val.object_range().begin(), val.object_range().end());
|
||||
} else if (val.is_array()) {
|
||||
val.erase(val.array_range().begin(), val.array_range().end());
|
||||
} else if (val.is_number()) {
|
||||
val = 0;
|
||||
if (val->is_object()) {
|
||||
val->erase(val->object_range().begin(), val->object_range().end());
|
||||
} else if (val->is_array()) {
|
||||
val->erase(val->array_range().begin(), val->array_range().end());
|
||||
} else if (val->is_number()) {
|
||||
*val = 0;
|
||||
}
|
||||
|
||||
clear_items += 1;
|
||||
return false;
|
||||
};
|
||||
|
||||
OpStatus status = UpdateEntry(op_args, key, path, cb);
|
||||
|
|
@ -832,19 +856,15 @@ OpResult<vector<OptString>> OpArrPop(const OpArgs& op_args, string_view key, str
|
|||
int index, JsonPathV2 expression) {
|
||||
vector<OptString> vec;
|
||||
OpStatus status;
|
||||
if (std::holds_alternative<json::Path>(expression)) {
|
||||
auto cb = [&vec, index](optional<string_view>, JsonType* val) {
|
||||
ArrayPop(nullopt, index, val, &vec);
|
||||
return false;
|
||||
};
|
||||
|
||||
if (holds_alternative<json::Path>(expression)) {
|
||||
const json::Path& json_path = std::get<json::Path>(expression);
|
||||
auto cb = [&vec, index](optional<string_view>, JsonType* val) {
|
||||
ArrayPop(nullopt, index, val, &vec);
|
||||
return false;
|
||||
};
|
||||
status = UpdateEntry(op_args, key, json_path, std::move(cb));
|
||||
} else {
|
||||
auto cb = [&](const auto& path, JsonType& val) {
|
||||
ArrayPop(nullopt, index, &val, &vec);
|
||||
return false;
|
||||
};
|
||||
|
||||
status = UpdateEntry(op_args, key, path, cb);
|
||||
}
|
||||
if (status != OpStatus::OK) {
|
||||
|
|
@ -857,15 +877,15 @@ OpResult<vector<OptString>> OpArrPop(const OpArgs& op_args, string_view key, str
|
|||
OpResult<vector<OptSizeT>> OpArrTrim(const OpArgs& op_args, string_view key, string_view path,
|
||||
int start_index, int stop_index) {
|
||||
vector<OptSizeT> vec;
|
||||
auto cb = [&](const auto&, JsonType& val) {
|
||||
if (!val.is_array()) {
|
||||
auto cb = [&](const auto&, JsonType* val) {
|
||||
if (!val->is_array()) {
|
||||
vec.emplace_back(nullopt);
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
if (val.empty()) {
|
||||
if (val->empty()) {
|
||||
vec.emplace_back(0);
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
size_t trim_start_index;
|
||||
|
|
@ -876,26 +896,27 @@ OpResult<vector<OptSizeT>> OpArrTrim(const OpArgs& op_args, string_view key, str
|
|||
}
|
||||
|
||||
size_t trim_end_index;
|
||||
if ((size_t)stop_index >= val.size()) {
|
||||
trim_end_index = val.size();
|
||||
if ((size_t)stop_index >= val->size()) {
|
||||
trim_end_index = val->size();
|
||||
} else {
|
||||
trim_end_index = stop_index;
|
||||
}
|
||||
|
||||
if (trim_start_index >= val.size() || trim_start_index > trim_end_index) {
|
||||
val.erase(val.array_range().begin(), val.array_range().end());
|
||||
if (trim_start_index >= val->size() || trim_start_index > trim_end_index) {
|
||||
val->erase(val->array_range().begin(), val->array_range().end());
|
||||
vec.emplace_back(0);
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
auto trim_start_it = std::next(val.array_range().begin(), trim_start_index);
|
||||
auto trim_end_it = val.array_range().end();
|
||||
if (trim_end_index < val.size()) {
|
||||
trim_end_it = std::next(val.array_range().begin(), trim_end_index + 1);
|
||||
auto trim_start_it = std::next(val->array_range().begin(), trim_start_index);
|
||||
auto trim_end_it = val->array_range().end();
|
||||
if (trim_end_index < val->size()) {
|
||||
trim_end_it = std::next(val->array_range().begin(), trim_end_index + 1);
|
||||
}
|
||||
|
||||
val = json_array<JsonType>(trim_start_it, trim_end_it);
|
||||
vec.emplace_back(val.size());
|
||||
*val = json_array<JsonType>(trim_start_it, trim_end_it);
|
||||
vec.emplace_back(val->size());
|
||||
return false;
|
||||
};
|
||||
|
||||
OpStatus status = UpdateEntry(op_args, key, path, cb);
|
||||
|
|
@ -913,46 +934,47 @@ OpResult<vector<OptSizeT>> OpArrInsert(const OpArgs& op_args, string_view key, s
|
|||
// Insert user-supplied value into the supplied index that should be valid.
|
||||
// If at least one index isn't valid within an array in the json doc, the operation is discarded.
|
||||
// Negative indexes start from the end of the array.
|
||||
auto cb = [&](const auto&, JsonType& val) {
|
||||
auto cb = [&](const auto&, JsonType* val) {
|
||||
if (out_of_boundaries_encountered) {
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!val.is_array()) {
|
||||
if (!val->is_array()) {
|
||||
vec.emplace_back(nullopt);
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
size_t removal_index;
|
||||
if (index < 0) {
|
||||
if (val.empty()) {
|
||||
if (val->empty()) {
|
||||
out_of_boundaries_encountered = true;
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
int temp_index = index + val.size();
|
||||
int temp_index = index + val->size();
|
||||
if (temp_index < 0) {
|
||||
out_of_boundaries_encountered = true;
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
removal_index = temp_index;
|
||||
} else {
|
||||
if ((size_t)index > val.size()) {
|
||||
if ((size_t)index > val->size()) {
|
||||
out_of_boundaries_encountered = true;
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
|
||||
removal_index = index;
|
||||
}
|
||||
|
||||
auto it = next(val.array_range().begin(), removal_index);
|
||||
auto it = next(val->array_range().begin(), removal_index);
|
||||
for (auto& new_val : new_values) {
|
||||
it = val.insert(it, new_val);
|
||||
it = val->insert(it, new_val);
|
||||
it++;
|
||||
}
|
||||
|
||||
vec.emplace_back(val.size());
|
||||
vec.emplace_back(val->size());
|
||||
return false;
|
||||
};
|
||||
|
||||
OpStatus status = UpdateEntry(op_args, key, path, cb);
|
||||
|
|
@ -978,18 +1000,19 @@ OpResult<vector<OptSizeT>> OpArrAppend(const OpArgs& op_args, string_view key, s
|
|||
return result.status();
|
||||
}
|
||||
|
||||
auto cb = [&](const auto&, JsonType& val) {
|
||||
if (!val.is_array()) {
|
||||
auto cb = [&](const auto&, JsonType* val) {
|
||||
if (!val->is_array()) {
|
||||
vec.emplace_back(nullopt);
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
for (auto& new_val : append_values) {
|
||||
val.emplace_back(new_val);
|
||||
val->emplace_back(new_val);
|
||||
}
|
||||
vec.emplace_back(val.size());
|
||||
vec.emplace_back(val->size());
|
||||
return false;
|
||||
};
|
||||
|
||||
OpStatus status = UpdateEntry(op_args, key, path, cb);
|
||||
OpStatus status = UpdateEntry(op_args, key, path, std::move(cb));
|
||||
if (status != OpStatus::OK) {
|
||||
return status;
|
||||
}
|
||||
|
|
@ -1194,12 +1217,13 @@ OpResult<bool> OpSet(const OpArgs& op_args, string_view key, string_view path,
|
|||
bool path_exists = false;
|
||||
bool operation_result = false;
|
||||
const JsonType& new_json = parsed_json.value();
|
||||
auto cb = [&](const auto&, JsonType& val) {
|
||||
auto cb = [&](const auto&, JsonType* val) {
|
||||
path_exists = true;
|
||||
if (!is_nx_condition) {
|
||||
operation_result = true;
|
||||
val = new_json;
|
||||
*val = new_json;
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
auto inserter = [&](JsonType& json) {
|
||||
|
|
@ -1287,7 +1311,7 @@ void JsonFamily::Set(CmdArgList args, ConnectionContext* cntx) {
|
|||
auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
|
||||
if (result) {
|
||||
if (*result) {
|
||||
rb->SendSimpleString("OK");
|
||||
rb->SendOk();
|
||||
} else {
|
||||
rb->SendNull();
|
||||
}
|
||||
|
|
@ -1331,9 +1355,9 @@ void JsonFamily::Debug(CmdArgList args, ConnectionContext* cntx) {
|
|||
if (absl::EqualsIgnoreCase(command, "help")) {
|
||||
auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
|
||||
rb->StartArray(2);
|
||||
rb->SendSimpleString(
|
||||
rb->SendBulkString(
|
||||
"JSON.DEBUG FIELDS <key> <path> - report number of fields in the JSON element.");
|
||||
rb->SendSimpleString("JSON.DEBUG HELP - print help message.");
|
||||
rb->SendBulkString("JSON.DEBUG HELP - print help message.");
|
||||
return;
|
||||
|
||||
} else if (absl::EqualsIgnoreCase(command, "fields")) {
|
||||
|
|
@ -1669,12 +1693,16 @@ void JsonFamily::ObjKeys(CmdArgList args, ConnectionContext* cntx) {
|
|||
void JsonFamily::Del(CmdArgList args, ConnectionContext* cntx) {
|
||||
string_view key = ArgS(args, 0);
|
||||
string_view path;
|
||||
|
||||
optional<JsonPathV2> expression;
|
||||
|
||||
if (args.size() > 1) {
|
||||
path = ArgS(args, 1);
|
||||
expression.emplace(PARSE_PATHV2(path));
|
||||
}
|
||||
|
||||
auto cb = [&](Transaction* t, EngineShard* shard) {
|
||||
return OpDel(t->GetOpArgs(shard), key, path);
|
||||
return OpDel(t->GetOpArgs(shard), key, path, std::move(expression));
|
||||
};
|
||||
|
||||
Transaction* trans = cntx->transaction;
|
||||
|
|
@ -1693,17 +1721,19 @@ void JsonFamily::NumIncrBy(CmdArgList args, ConnectionContext* cntx) {
|
|||
return;
|
||||
}
|
||||
|
||||
JsonPathV2 expression = PARSE_PATHV2(path);
|
||||
|
||||
auto cb = [&](Transaction* t, EngineShard* shard) {
|
||||
return OpDoubleArithmetic(t->GetOpArgs(shard), key, path, dnum, plus<double>{});
|
||||
return OpDoubleArithmetic(t->GetOpArgs(shard), key, path, dnum, OP_ADD, std::move(expression));
|
||||
};
|
||||
|
||||
Transaction* trans = cntx->transaction;
|
||||
OpResult<string> result = trans->ScheduleSingleHopT(std::move(cb));
|
||||
OpResult<string> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
|
||||
auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
|
||||
|
||||
if (result) {
|
||||
cntx->SendSimpleString(*result);
|
||||
rb->SendBulkString(*result);
|
||||
} else {
|
||||
cntx->SendError(result.status());
|
||||
rb->SendError(result.status());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1718,17 +1748,20 @@ void JsonFamily::NumMultBy(CmdArgList args, ConnectionContext* cntx) {
|
|||
return;
|
||||
}
|
||||
|
||||
JsonPathV2 expression = PARSE_PATHV2(path);
|
||||
|
||||
auto cb = [&](Transaction* t, EngineShard* shard) {
|
||||
return OpDoubleArithmetic(t->GetOpArgs(shard), key, path, dnum, multiplies<double>{});
|
||||
return OpDoubleArithmetic(t->GetOpArgs(shard), key, path, dnum, OP_MULTIPLY,
|
||||
std::move(expression));
|
||||
};
|
||||
|
||||
Transaction* trans = cntx->transaction;
|
||||
OpResult<string> result = trans->ScheduleSingleHopT(std::move(cb));
|
||||
auto* rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
|
||||
OpResult<string> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
|
||||
|
||||
if (result) {
|
||||
cntx->SendSimpleString(*result);
|
||||
rb->SendBulkString(*result);
|
||||
} else {
|
||||
cntx->SendError(result.status());
|
||||
rb->SendError(result.status());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -363,8 +363,11 @@ TEST_F(JsonFamilyTest, NumIncrBy) {
|
|||
resp = Run({"JSON.NUMINCRBY", "json", "$.d[*]", "1"});
|
||||
EXPECT_EQ(resp, "[2,3,4]");
|
||||
|
||||
resp = Run({"JSON.NUMINCRBY", "json", "$.d[2]", "1"});
|
||||
EXPECT_EQ(resp, "[5]");
|
||||
|
||||
resp = Run({"JSON.GET", "json", "$.*"});
|
||||
EXPECT_EQ(resp, R"([[],[2],[2,3],[2,3,4]])");
|
||||
EXPECT_EQ(resp, R"([[],[2],[2,3],[2,3,5]])");
|
||||
|
||||
json = R"(
|
||||
{"a":{}, "b":{"a":1}, "c":{"a":1, "b":2}, "d":{"a":1, "b":2, "c":3}}
|
||||
|
|
@ -513,7 +516,11 @@ TEST_F(JsonFamilyTest, Del) {
|
|||
EXPECT_EQ(resp, R"({"a":{},"b":{"a":1},"c":{"a":1,"b":2},"d":{},"e":[]})");
|
||||
|
||||
resp = Run({"JSON.DEL", "json", "$..*"});
|
||||
EXPECT_THAT(resp, IntArg(8));
|
||||
|
||||
// TODO: legacy jsoncons implementation returns, 8 but in practive it should return 5.
|
||||
// redis-stack returns 5 as well.
|
||||
// Once we drop jsoncons path, we can enforce here equality.
|
||||
EXPECT_GE(resp.GetInt(), 5);
|
||||
|
||||
resp = Run({"JSON.GET", "json"});
|
||||
EXPECT_EQ(resp, R"({})");
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import dataclasses
|
||||
import os
|
||||
import time
|
||||
import subprocess
|
||||
import aiohttp
|
||||
|
|
@ -190,7 +191,8 @@ class DflyInstance:
|
|||
self._port = None
|
||||
|
||||
all_args = self.format_args(self.args)
|
||||
logging.debug(f"Starting instance with arguments {all_args} from {self.params.path}")
|
||||
real_path = os.path.realpath(self.params.path)
|
||||
logging.debug(f"Starting instance with arguments {' '.join(all_args)} from {real_path}")
|
||||
|
||||
run_cmd = [self.params.path, *all_args]
|
||||
if self.params.gdb:
|
||||
|
|
@ -315,6 +317,8 @@ class DflyInstanceFactory:
|
|||
def create(self, existing_port=None, **kwargs) -> DflyInstance:
|
||||
args = {**self.args, **kwargs}
|
||||
args.setdefault("dbfilename", "")
|
||||
args.setdefault("jsonpathv2", None)
|
||||
|
||||
vmod = "dragonfly_connection=1,accept_server=1,listener_interface=1,main_service=1,rdb_save=1,replica=1,cluster_family=1"
|
||||
args.setdefault("vmodule", vmod)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue