Skip to content

Commit

Permalink
chore: fully support qlist in all list_family commands (dragonflydb#4201
Browse files Browse the repository at this point in the history 
)

Signed-off-by: Roman Gershman <[email protected]>
  • Loading branch information
@romange
romange authored Nov 27, 2024
1 parent 66e0fd0 commit a87d6ae
Show file tree
Hide file tree
Showing 2 changed files with 104 additions and 57 deletions.
3 changes: 2 additions & 1 deletion .github/workflows/ci.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,8 @@ jobs:
run: |
cd ${GITHUB_WORKSPACE}/build
echo Run ctest -V -L DFLY
GLOG_alsologtostderr=1 GLOG_vmodule=rdb_load=1,rdb_save=1,snapshot=1 timeout 20m ctest -V -L DFLY
GLOG_alsologtostderr=1 GLOG_vmodule=rdb_load=1,rdb_save=1,snapshot=1 \
FLAGS_list_experimental_v2=true timeout 20m ctest -V -L DFLY
echo "Running tests with --force_epoll"
Expand Down
158 changes: 102 additions & 56 deletions src/server/list_family.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -571,47 +571,75 @@ OpResult<string> OpIndex(const OpArgs& op_args, std::string_view key, long index
}

OpResult<vector<uint32_t>> OpPos(const OpArgs& op_args, string_view key, string_view element,
int rank, int count, int max_len) {
int rank, uint32_t count, uint32_t max_len) {
DCHECK(key.data() && element.data());
DCHECK_NE(rank, 0);

auto it_res = op_args.GetDbSlice().FindReadOnly(op_args.db_cntx, key, OBJ_LIST);
if (!it_res.ok())
return it_res.status();

int direction = AL_START_HEAD;
if (rank < 0) {
rank = -rank;
direction = AL_START_TAIL;
}

quicklist* ql = GetQL(it_res.value()->second);
quicklistIter* ql_iter = quicklistGetIterator(ql, direction);
quicklistEntry entry;

int index = 0;
int matched = 0;
const PrimeValue& pv = (*it_res)->second;
vector<uint32_t> matches;
string str;

while (quicklistNext(ql_iter, &entry) && (max_len == 0 || index < max_len)) {
if (entry.value) {
str.assign(reinterpret_cast<char*>(entry.value), entry.sz);
} else {
str = absl::StrCat(entry.longval);
if (pv.Encoding() == kEncodingQL2) {
QList* ql = GetQLV2(pv);
QList::Where where = QList::HEAD;
if (rank < 0) {
rank = -rank;
where = QList::TAIL;
}
if (str == element) {
matched++;
auto k = (direction == AL_START_TAIL) ? ql->count - index - 1 : index;
if (matched >= rank) {
matches.push_back(k);
if (count && matched - rank + 1 >= count) {
break;

auto it = ql->GetIterator(where);
unsigned index = 0;
while (it.Next() && (max_len == 0 || index < max_len)) {
if (it.Get() == element) {
if (rank == 1) {
auto k = (where == QList::HEAD) ? index : ql->Size() - index - 1;
matches.push_back(k);
if (count && matches.size() >= count)
break;
} else {
rank--;
}
}
index++;
}
} else {
int direction = AL_START_HEAD;
if (rank < 0) {
rank = -rank;
direction = AL_START_TAIL;
}

quicklist* ql = GetQL(it_res.value()->second);
quicklistIter* ql_iter = quicklistGetIterator(ql, direction);
quicklistEntry entry;

unsigned index = 0;
int matched = 0;
string str;

while (quicklistNext(ql_iter, &entry) && (max_len == 0 || index < max_len)) {
if (entry.value) {
str.assign(reinterpret_cast<char*>(entry.value), entry.sz);
} else {
str = absl::StrCat(entry.longval);
}
if (str == element) {
matched++;
auto k = (direction == AL_START_TAIL) ? ql->count - index - 1 : index;
if (matched >= rank) {
matches.push_back(k);
if (count && unsigned(matched - rank + 1) >= count) {
break;
}
}
}
index++;
}
index++;
quicklistReleaseIterator(ql_iter);
}
quicklistReleaseIterator(ql_iter);
return matches;
}

Expand All @@ -624,29 +652,41 @@ OpResult<int> OpInsert(const OpArgs& op_args, string_view key, string_view pivot
if (!it_res)
return it_res.status();

quicklist* ql = GetQL(it_res->it->second);
quicklistEntry entry = container_utils::QLEntry();
quicklistIter* qiter = quicklistGetIterator(ql, AL_START_HEAD);
bool found = false;
PrimeValue& pv = it_res->it->second;

while (quicklistNext(qiter, &entry)) {
if (ElemCompare(entry, pivot)) {
found = true;
break;
int res = -1;

if (pv.Encoding() == kEncodingQL2) {
QList* ql = GetQLV2(pv);
QList::InsertOpt insert_opt = (insert_param == INSERT_BEFORE) ? QList::BEFORE : QList::AFTER;
if (ql->Insert(pivot, elem, insert_opt)) {
res = ql->Size();
}
}
} else {
quicklist* ql = GetQL(pv);
quicklistEntry entry = container_utils::QLEntry();
quicklistIter* qiter = quicklistGetIterator(ql, AL_START_HEAD);
bool found = false;

int res = -1;
if (found) {
if (insert_param == INSERT_AFTER) {
quicklistInsertAfter(qiter, &entry, elem.data(), elem.size());
} else {
DCHECK_EQ(INSERT_BEFORE, insert_param);
quicklistInsertBefore(qiter, &entry, elem.data(), elem.size());
while (quicklistNext(qiter, &entry)) {
if (ElemCompare(entry, pivot)) {
found = true;
break;
}
}

if (found) {
if (insert_param == INSERT_AFTER) {
quicklistInsertAfter(qiter, &entry, elem.data(), elem.size());
} else {
DCHECK_EQ(INSERT_BEFORE, insert_param);
quicklistInsertBefore(qiter, &entry, elem.data(), elem.size());
}
res = quicklistCount(ql);
}
res = quicklistCount(ql);
quicklistReleaseIterator(qiter);
}
quicklistReleaseIterator(qiter);

return res;
}

Expand Down Expand Up @@ -738,14 +778,21 @@ OpStatus OpSet(const OpArgs& op_args, string_view key, string_view elem, long in
return it_res.status();

auto it = it_res->it;
quicklist* ql = GetQL(it->second);

int replaced = quicklistReplaceAtIndex(ql, index, elem.data(), elem.size());
OpStatus status = OpStatus::OUT_OF_RANGE;
if (it->second.Encoding() == kEncodingQL2) {
QList* ql = GetQLV2(it->second);
if (ql->Replace(index, elem))
status = OpStatus::OK;
} else {
DCHECK_EQ(it->second.Encoding(), OBJ_ENCODING_QUICKLIST);
quicklist* ql = GetQL(it->second);

if (!replaced) {
return OpStatus::OUT_OF_RANGE;
int replaced = quicklistReplaceAtIndex(ql, index, elem.data(), elem.size());
if (replaced) {
status = OpStatus::OK;
}
}
return OpStatus::OK;
return status;
}

OpStatus OpTrim(const OpArgs& op_args, string_view key, long start, long end) {
Expand Down Expand Up @@ -803,8 +850,8 @@ OpResult<StringVec> OpRange(const OpArgs& op_args, std::string_view key, long st
if (!res)
return res.status();

quicklist* ql = GetQL(res.value()->second);
long llen = quicklistCount(ql);
const PrimeValue& pv = (*res)->second;
long llen = pv.Size();

/* convert negative indexes */
if (start < 0)
Expand All @@ -823,13 +870,12 @@ OpResult<StringVec> OpRange(const OpArgs& op_args, std::string_view key, long st

StringVec str_vec;
container_utils::IterateList(
res.value()->second,
pv,
[&str_vec](container_utils::ContainerEntry ce) {
str_vec.emplace_back(ce.ToString());
return true;
},
start, end);

return str_vec;
}

Expand Down

0 comments on commit a87d6ae

Please sign in to comment.