mirrors/scylladb
1
0
Fork 0
mirror of https://github.com/scylladb/scylladb.git synced 2026-05-28 10:41:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

mvcc: partition_snapshot_row_cursor: Fix cursor skipping over rows added after its position

Browse Source 
The cursor maintains a heap of iterators in all versions. If rows were
inserted before the latest version's iterator, cursor would not see
them. Fix by redoing the lookup for iterators not in the current row
in maybe_refresh().

Refs #2834.
This commit is contained in:
Tomasz Grabiec
2017-09-21 18:07:16 +02:00
parent 09d99b0358
commit 2f8d91043d
1 changed files with 33 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
partition_snapshot_row_cursor.hh
View File

@@ -34,6 +34,8 @@
// When the cursor is invalidated, it still maintains its previous position. It can be brought
// back to validity by calling maybe_refresh(), or advance_to().
//
// Insertion of row entries after cursor's position invalidates the cursor.
//
class partition_snapshot_row_cursor final {
struct position_in_version {
mutation_partition::rows_type::iterator it;
@@ -101,6 +103,37 @@ public:
if (!iterators_valid()) {
return advance_to(_position);
}
// Refresh latest version's iterator in case there was an insertion
// before it and after cursor's position. There cannot be any
// insertions for non-latest versions, so we don't have to update them.
if (_current_row[0].version_no != 0) {
rows_entry::compare less(_schema);
position_in_partition::equal_compare eq(_schema);
position_in_version::less_compare heap_less(_schema);
auto& rows = _snp.version()->partition().clustered_rows();
auto it = _iterators[0] = rows.lower_bound(_position, less);
auto heap_i = boost::find_if(_heap, [](auto&& v) { return v.version_no == 0; });
if (it == rows.end()) {
if (heap_i != _heap.end()) {
_heap.erase(heap_i);
boost::range::make_heap(_heap, heap_less);
}
} else if (eq(_position, it->position())) {
_current_row.insert(_current_row.begin(), position_in_version{it, rows.end(), 0});
if (heap_i != _heap.end()) {
_heap.erase(heap_i);
boost::range::make_heap(_heap, heap_less);
}
} else {
if (heap_i != _heap.end()) {
heap_i->it = it;
boost::range::make_heap(_heap, heap_less);
} else {
_heap.push_back({it, rows.end(), 0});
boost::range::push_heap(_heap, heap_less);
}
}
}
return true;
}