You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by ad...@apache.org on 2018/07/25 20:30:18 UTC
[1/2] kudu git commit: memrowset: support iteration with
snap_to_exclude
Repository: kudu
Updated Branches:
refs/heads/master 524485ebc -> 0418b225e
memrowset: support iteration with snap_to_exclude
Following on the commit that introduced the snap_to_exclude iterator option,
this commit modifies the MemRowSet iterator to take snap_to_exclude into
consideration (if it is set) when determining which rows and mutations are
relevant to an iterator.
I've also pulled the "exceeded the iterator's scan bounds" short-circuit out
of the MVCC checks. If it were true for a row excluded by MVCC, it'd be true
for all subsequent rows too.
Change-Id: I9be2ab5af4d5223889e4545ae1db7cc0275480ce
Reviewed-on: http://gerrit.cloudera.org:8080/10926
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: Kudu Jenkins
Reviewed-by: Mike Percy <mp...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/5ba57a07
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/5ba57a07
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/5ba57a07
Branch: refs/heads/master
Commit: 5ba57a0728dad13bd0caa1ea5621f7e122497284
Parents: 524485e
Author: Adar Dembo <ad...@cloudera.com>
Authored: Tue Jul 10 16:25:31 2018 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Wed Jul 25 20:29:37 2018 +0000
----------------------------------------------------------------------
src/kudu/tablet/diskrowset-test.cc | 8 ++-
src/kudu/tablet/memrowset-test.cc | 88 +++++++++++++++++++++++++++++++--
src/kudu/tablet/memrowset.cc | 70 +++++++++++++++++++-------
src/kudu/tablet/memrowset.h | 28 +++++++++--
src/kudu/tablet/tablet-test-util.h | 19 +++----
5 files changed, 173 insertions(+), 40 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/5ba57a07/src/kudu/tablet/diskrowset-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/diskrowset-test.cc b/src/kudu/tablet/diskrowset-test.cc
index 105e8ba..ce0fb7d 100644
--- a/src/kudu/tablet/diskrowset-test.cc
+++ b/src/kudu/tablet/diskrowset-test.cc
@@ -272,15 +272,19 @@ TEST_F(TestRowSet, TestDelete) {
vector<string> rows;
Status s;
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
for (int i = 0; i < 2; i++) {
// Reading the MVCC snapshot prior to deletion should show the row.
- ASSERT_OK(DumpRowSet(*rs, schema_, snap_before_delete, &rows));
+ opts.snap_to_include = snap_before_delete;
+ ASSERT_OK(DumpRowSet(*rs, opts, &rows));
ASSERT_EQ(2, rows.size());
EXPECT_EQ(R"((string key="hello 000000000000000", uint32 val=0))", rows[0]);
EXPECT_EQ(R"((string key="hello 000000000000001", uint32 val=1))", rows[1]);
// Reading the MVCC snapshot after the deletion should hide the row.
- ASSERT_OK(DumpRowSet(*rs, schema_, snap_after_delete, &rows));
+ opts.snap_to_include = snap_after_delete;
+ ASSERT_OK(DumpRowSet(*rs, opts, &rows));
ASSERT_EQ(1, rows.size());
EXPECT_EQ(R"((string key="hello 000000000000001", uint32 val=1))", rows[0]);
http://git-wip-us.apache.org/repos/asf/kudu/blob/5ba57a07/src/kudu/tablet/memrowset-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset-test.cc b/src/kudu/tablet/memrowset-test.cc
index 027d05e..7587694 100644
--- a/src/kudu/tablet/memrowset-test.cc
+++ b/src/kudu/tablet/memrowset-test.cc
@@ -24,6 +24,7 @@
#include <string>
#include <vector>
+#include <boost/optional/optional.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <gtest/gtest.h>
@@ -438,17 +439,22 @@ TEST_F(TestMemRowSet, TestDelete) {
rows[0]);
// Verify that iterating the rowset at the first snapshot shows the row.
- ASSERT_OK(DumpRowSet(*mrs, schema_, snapshot_before_delete, &rows));
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
+ opts.snap_to_include = snapshot_before_delete;
+ ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
ASSERT_EQ(1, rows.size());
EXPECT_EQ(R"((string key="hello world", uint32 val=1))", rows[0]);
// Verify that iterating the rowset at the snapshot where it's deleted
// doesn't show the row.
- ASSERT_OK(DumpRowSet(*mrs, schema_, snapshot_after_delete, &rows));
+ opts.snap_to_include = snapshot_after_delete;
+ ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
ASSERT_EQ(0, rows.size());
// Verify that iterating the rowset after it's re-inserted shows the row.
- ASSERT_OK(DumpRowSet(*mrs, schema_, snapshot_after_reinsert, &rows));
+ opts.snap_to_include = snapshot_after_reinsert;
+ ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
ASSERT_EQ(1, rows.size());
EXPECT_EQ(R"((string key="hello world", uint32 val=2))", rows[0]);
}
@@ -511,11 +517,14 @@ TEST_F(TestMemRowSet, TestInsertionMVCC) {
ASSERT_OK(mrs->DebugDump(nullptr));
ASSERT_EQ(5, snapshots.size());
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
for (int i = 0; i < 5; i++) {
SCOPED_TRACE(i);
// Each snapshot 'i' is taken after row 'i' was committed.
+ opts.snap_to_include = snapshots[i];
vector<string> rows;
- ASSERT_OK(kudu::tablet::DumpRowSet(*mrs, schema_, snapshots[i], &rows));
+ ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
ASSERT_EQ(1 + i, rows.size());
string expected = StringPrintf(R"((string key="tx%d", uint32 val=%d))", i, i);
ASSERT_EQ(expected, rows[i]);
@@ -552,10 +561,13 @@ TEST_F(TestMemRowSet, TestUpdateMVCC) {
// Validate that each snapshot returns the expected value
ASSERT_EQ(6, snapshots.size());
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
for (int i = 0; i <= 5; i++) {
SCOPED_TRACE(i);
vector<string> rows;
- ASSERT_OK(kudu::tablet::DumpRowSet(*mrs, schema_, snapshots[i], &rows));
+ opts.snap_to_include = snapshots[i];
+ ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
ASSERT_EQ(1, rows.size());
string expected = StringPrintf(R"((string key="my row", uint32 val=%d))", i);
@@ -565,5 +577,71 @@ TEST_F(TestMemRowSet, TestUpdateMVCC) {
}
}
+TEST_F(TestMemRowSet, TestScanSnapToExclude) {
+ shared_ptr<MemRowSet> mrs;
+ ASSERT_OK(MemRowSet::Create(0, schema_, log_anchor_registry_.get(),
+ MemTracker::GetRootTracker(), &mrs));
+
+ // Sequence of operations:
+ // @1
+ // @2 INSERT key=row val=0
+ // @3 UPDATE key=row val=1
+ // @4 DELETE key=row
+ // @5 INSERT key=row val=2
+ vector<MvccSnapshot> snaps;
+ snaps.emplace_back(mvcc_);
+ ASSERT_OK(InsertRow(mrs.get(), "row", 0));
+ snaps.emplace_back(mvcc_);
+ OperationResultPB result;
+ ASSERT_OK(UpdateRow(mrs.get(), "row", 1, &result));
+ snaps.emplace_back(mvcc_);
+ ASSERT_OK(DeleteRow(mrs.get(), "row", &result));
+ snaps.emplace_back(mvcc_);
+ ASSERT_OK(InsertRow(mrs.get(), "row", 2));
+ snaps.emplace_back(mvcc_);
+
+ auto DumpAndCheck = [&](const MvccSnapshot& exclude,
+ const MvccSnapshot& include,
+ boost::optional<int> row_value = boost::none) {
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
+ opts.snap_to_include = include;
+ opts.snap_to_exclude = exclude;
+ vector<string> rows;
+ ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
+ if (row_value.is_initialized()) {
+ ASSERT_EQ(1, rows.size());
+ ASSERT_EQ(StringPrintf(R"((string key="row", uint32 val=%d))", row_value.get()), rows[0]);
+ } else {
+ ASSERT_TRUE(rows.empty());
+ }
+ };
+
+ // Captures zero rows; a snapshot range [x, x) does not include anything.
+ for (const auto& s : snaps) {
+ NO_FATALS(DumpAndCheck(s, s));
+ }
+
+ {
+ NO_FATALS(DumpAndCheck(snaps[0], snaps[1], 0)); // INSERT
+ NO_FATALS(DumpAndCheck(snaps[1], snaps[2], 1)); // UPDATE
+ NO_FATALS(DumpAndCheck(snaps[2], snaps[3])); // DELETE
+ NO_FATALS(DumpAndCheck(snaps[3], snaps[4], 2)); // REINSERT
+ }
+
+ {
+ NO_FATALS(DumpAndCheck(snaps[0], snaps[2], 1)); // INSERT, UPDATE
+ NO_FATALS(DumpAndCheck(snaps[1], snaps[3])); // UPDATE, DELETE
+ NO_FATALS(DumpAndCheck(snaps[2], snaps[4], 2)); // DELETE, REINSERT
+ }
+
+ {
+ NO_FATALS(DumpAndCheck(snaps[0], snaps[3])); // INSERT, UPDATE, DELETE
+ NO_FATALS(DumpAndCheck(snaps[1], snaps[4], 2)); // UPDATE, DELETE, REINSERT
+ }
+
+ NO_FATALS(DumpAndCheck(snaps[0], snaps[4], 2)); // INSERT, UPDATE, DELETE, REINSERT
+}
+
} // namespace tablet
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/5ba57a07/src/kudu/tablet/memrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc
index 5e90eb9..213b7b9 100644
--- a/src/kudu/tablet/memrowset.cc
+++ b/src/kudu/tablet/memrowset.cc
@@ -496,21 +496,44 @@ Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) {
iter_->GetCurrentEntry(&k, &v);
MRSRow row(memrowset_.get(), v);
- if (opts_.snap_to_include.IsCommitted(row.insertion_timestamp())) {
- if (has_upper_bound() && out_of_bounds(k)) {
- state_ = kFinished;
- break;
- } else {
- RETURN_NOT_OK(projector_->ProjectRowForRead(row, &dst_row, dst->arena()));
+ // Short-circuit if we've exceeded the iteration's upper bound.
+ if (has_upper_bound() && out_of_bounds(k)) {
+ state_ = kFinished;
+ break;
+ }
- Mutation* redo_head = reinterpret_cast<Mutation*>(
- base::subtle::Acquire_Load(reinterpret_cast<AtomicWord*>(&row.header_->redo_head)));
- // Roll-forward MVCC for committed updates.
- RETURN_NOT_OK(ApplyMutationsToProjectedRow(
- redo_head, &dst_row, dst->arena()));
- }
+ // The snapshots in 'opts_' represent a time range that this iterator must
+ // respect. There are two possible cases:
+ //
+ // 1. 'snap_to_exclude' is unset but 'snap_to_include' is set. The time
+ // range is [INF, snap_to_include).
+ // 2. Both 'snap_to exclude' and 'snap_to_include' are set. The time range
+ // is [snap_to_exclude, snap_to_include).
+ //
+ // If the row's insertion timestamp is committed in 'snap_to_exclude', it
+ // means the insertion was outside this iterator's time range (i.e. the
+ // insert was "excluded"). However, subsequent mutations may be inside the
+ // time range, so we must still project the row and walk its mutation list.
+ bool insert_excluded = opts_.snap_to_exclude &&
+ opts_.snap_to_exclude->IsCommitted(row.insertion_timestamp());
+ bool unset_in_sel_vector;
+ if (insert_excluded || opts_.snap_to_include.IsCommitted(row.insertion_timestamp())) {
+ RETURN_NOT_OK(projector_->ProjectRowForRead(row, &dst_row, dst->arena()));
+
+ // Roll-forward MVCC for committed updates.
+ Mutation* redo_head = reinterpret_cast<Mutation*>(
+ base::subtle::Acquire_Load(reinterpret_cast<AtomicWord*>(&row.header_->redo_head)));
+ ApplyStatus apply_status;
+ RETURN_NOT_OK(ApplyMutationsToProjectedRow(
+ redo_head, &dst_row, dst->arena(), &apply_status));
+ unset_in_sel_vector = apply_status == APPLIED_AND_DELETED ||
+ (apply_status == NONE_APPLIED && insert_excluded);
} else {
- // This row was not yet committed in the current MVCC snapshot
+ // The insertion is too new; the entire row should be omitted.
+ unset_in_sel_vector = true;
+ }
+
+ if (unset_in_sel_vector) {
dst->selection_vector()->SetRowUnselected(*fetched);
// In debug mode, fill the row data for easy debugging
@@ -530,10 +553,14 @@ Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) {
}
Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
- const Mutation *mutation_head, RowBlockRow *dst_row, Arena *dst_arena) {
+ const Mutation* mutation_head, RowBlockRow* dst_row, Arena* dst_arena,
+ ApplyStatus* apply_status) {
+ ApplyStatus local_apply_status = NONE_APPLIED;
+
// Fast short-circuit the likely case of a row which was inserted and never
// updated.
if (PREDICT_TRUE(mutation_head == nullptr)) {
+ *apply_status = local_apply_status;
return Status::OK();
}
@@ -543,10 +570,18 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
mut != nullptr;
mut = mut->acquire_next()) {
if (!opts_.snap_to_include.IsCommitted(mut->timestamp_)) {
- // Transaction which wasn't committed yet in the reader's snapshot.
+ // This mutation is too new; it should be omitted.
continue;
}
+ // If the mutation is too old, we still need to apply it (so that the column
+ // values are correct if we see a relevant mutation later), but it doesn't
+ // count towards the overall "application status".
+ if (!opts_.snap_to_exclude ||
+ !opts_.snap_to_exclude->IsCommitted(mut->timestamp_)) {
+ local_apply_status = APPLIED_AND_PRESENT;
+ }
+
// Apply the mutation.
// Check if it's a deletion.
@@ -575,10 +610,11 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
// If the most recent mutation seen for the row was a DELETE, then set the selection
// vector bit to 0, so it doesn't show up in the results.
- if (is_deleted) {
- dst_row->SetRowUnselected();
+ if (is_deleted && local_apply_status == APPLIED_AND_PRESENT) {
+ local_apply_status = APPLIED_AND_DELETED;
}
+ *apply_status = local_apply_status;
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/5ba57a07/src/kudu/tablet/memrowset.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.h b/src/kudu/tablet/memrowset.h
index aba1988..9b10187 100644
--- a/src/kudu/tablet/memrowset.h
+++ b/src/kudu/tablet/memrowset.h
@@ -542,11 +542,31 @@ class MemRowSet::Iterator : public RowwiseIterator {
Iterator(const std::shared_ptr<const MemRowSet> &mrs,
MemRowSet::MSBTIter *iter, RowIteratorOptions opts);
- // Various helper functions called while getting the next RowBlock
+ // Retrieves a block of dst->nrows() rows from the MemRowSet.
+ //
+ // Writes the number of rows retrieved to 'fetched'.
Status FetchRows(RowBlock* dst, size_t* fetched);
- Status ApplyMutationsToProjectedRow(const Mutation *mutation_head,
- RowBlockRow *dst_row,
- Arena *dst_arena);
+
+ // Walks the mutations in 'mutation_head', applying relevant ones to 'dst_row'
+ // (performing any allocations out of 'dst_arena').
+ //
+ // On success, 'apply_status' summarizes the application process.
+ enum ApplyStatus {
+ // No mutations were applied to the row, either because none existed or
+ // because none were relevant.
+ NONE_APPLIED,
+
+ // At least one mutation was applied to the row.
+ APPLIED_AND_PRESENT,
+
+ // At least one mutation was applied to the row, and the row's final state
+ // was deleted (i.e. the last mutation was a DELETE).
+ APPLIED_AND_DELETED,
+ };
+ Status ApplyMutationsToProjectedRow(const Mutation* mutation_head,
+ RowBlockRow* dst_row,
+ Arena* dst_arena,
+ ApplyStatus* apply_status);
const std::shared_ptr<const MemRowSet> memrowset_;
gscoped_ptr<MemRowSet::MSBTIter> iter_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/5ba57a07/src/kudu/tablet/tablet-test-util.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test-util.h b/src/kudu/tablet/tablet-test-util.h
index b45c75d..fa8d4f8 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -225,21 +225,16 @@ static inline void VerifySnapshotsHaveSameResult(
}
}
-// Construct a new iterator from the given rowset, and dump
-// all of its results into 'out'. The previous contents
-// of 'out' are cleared.
-static inline Status DumpRowSet(const RowSet &rs,
- const Schema &projection,
- const MvccSnapshot &snap,
- std::vector<std::string> *out,
- int limit = INT_MAX) {
- RowIteratorOptions opts;
- opts.projection = &projection;
- opts.snap_to_include = snap;
+// Constructs a new iterator for 'rs' with 'opts' and dumps all of its rows into 'out'.
+//
+// The previous contents of 'out' are cleared.
+static inline Status DumpRowSet(const RowSet& rs,
+ const RowIteratorOptions& opts,
+ std::vector<std::string>* out) {
gscoped_ptr<RowwiseIterator> iter;
RETURN_NOT_OK(rs.NewRowIterator(opts, &iter));
RETURN_NOT_OK(iter->Init(nullptr));
- RETURN_NOT_OK(IterateToStringList(iter.get(), out, limit));
+ RETURN_NOT_OK(IterateToStringList(iter.get(), out));
return Status::OK();
}
[2/2] kudu git commit: memrowset: support iteration with
include_deleted_rows
Posted by ad...@apache.org.
memrowset: support iteration with include_deleted_rows
This is another piece in the incremental backup puzzle. The idea is that
when taking an incremental backup, the scan will include rows that have been
deleted. The backup will figure out which rows were deleted by including a
special "is deleted" virtual column in the projection.
This commit introduces a new iterator option that can be used to include
deleted rows, and uses it in the MemRowSet.
Change-Id: Ie018043518b437ecc719cf9f87b2c5eea560c9a1
Reviewed-on: http://gerrit.cloudera.org:8080/10929
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: Kudu Jenkins
Reviewed-by: Mike Percy <mp...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/0418b225
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/0418b225
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/0418b225
Branch: refs/heads/master
Commit: 0418b225e93fd7eb2e1342850f559314d23a9a7b
Parents: 5ba57a0
Author: Adar Dembo <ad...@cloudera.com>
Authored: Thu Jul 12 11:54:16 2018 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Wed Jul 25 20:29:46 2018 +0000
----------------------------------------------------------------------
src/kudu/tablet/memrowset-test.cc | 86 ++++++++++++++++++++++++++++------
src/kudu/tablet/memrowset.cc | 2 +-
src/kudu/tablet/rowset.cc | 3 +-
src/kudu/tablet/rowset.h | 5 ++
4 files changed, 80 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/0418b225/src/kudu/tablet/memrowset-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset-test.cc b/src/kudu/tablet/memrowset-test.cc
index 7587694..627cbad 100644
--- a/src/kudu/tablet/memrowset-test.cc
+++ b/src/kudu/tablet/memrowset-test.cc
@@ -190,10 +190,7 @@ class TestMemRowSet : public KuduTest {
return s;
}
- int ScanAndCount(MemRowSet* mrs, const MvccSnapshot& snap) {
- RowIteratorOptions opts;
- opts.projection = &schema_;
- opts.snap_to_include = snap;
+ int ScanAndCount(MemRowSet* mrs, const RowIteratorOptions& opts) {
gscoped_ptr<MemRowSet::Iterator> iter(mrs->NewIterator(opts));
CHECK_OK(iter->Init(nullptr));
@@ -476,17 +473,20 @@ TEST_F(TestMemRowSet, TestMemRowSetInsertCountAndScan) {
}
for (int i = 0; i < FLAGS_num_scan_passes; i++) {
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
+ opts.snap_to_include = MvccSnapshot(Timestamp(0));
LOG_TIMING(INFO, "Scanning rows where none are committed") {
- ASSERT_EQ(0, ScanAndCount(mrs.get(), MvccSnapshot(Timestamp(0))));
+ ASSERT_EQ(0, ScanAndCount(mrs.get(), opts));
}
+ opts.snap_to_include = MvccSnapshot(Timestamp(FLAGS_roundtrip_num_rows + 1));
LOG_TIMING(INFO, "Scanning rows where all are committed") {
- ASSERT_EQ(FLAGS_roundtrip_num_rows,
- ScanAndCount(mrs.get(),
- MvccSnapshot(Timestamp(FLAGS_roundtrip_num_rows + 1))));
+ ASSERT_EQ(FLAGS_roundtrip_num_rows, ScanAndCount(mrs.get(), opts));
}
}
}
+
// Test that scanning at past MVCC snapshots will hide rows which are
// not committed in that snapshot.
TEST_F(TestMemRowSet, TestInsertionMVCC) {
@@ -577,7 +577,16 @@ TEST_F(TestMemRowSet, TestUpdateMVCC) {
}
}
-TEST_F(TestMemRowSet, TestScanSnapToExclude) {
+class ParameterizedTestMemRowSet : public TestMemRowSet,
+ public ::testing::WithParamInterface<bool> {
+};
+
+
+// Parameterized on opts.include_deleted_rows true or false.
+INSTANTIATE_TEST_CASE_P(IncludeDeletedRowsOrNot, ParameterizedTestMemRowSet,
+ ::testing::Bool());
+
+TEST_P(ParameterizedTestMemRowSet, TestScanSnapToExclude) {
shared_ptr<MemRowSet> mrs;
ASSERT_OK(MemRowSet::Create(0, schema_, log_anchor_registry_.get(),
MemTracker::GetRootTracker(), &mrs));
@@ -602,11 +611,12 @@ TEST_F(TestMemRowSet, TestScanSnapToExclude) {
auto DumpAndCheck = [&](const MvccSnapshot& exclude,
const MvccSnapshot& include,
- boost::optional<int> row_value = boost::none) {
+ boost::optional<int> row_value) {
RowIteratorOptions opts;
opts.projection = &schema_;
opts.snap_to_include = include;
opts.snap_to_exclude = exclude;
+ opts.include_deleted_rows = GetParam();
vector<string> rows;
ASSERT_OK(DumpRowSet(*mrs, opts, &rows));
if (row_value.is_initialized()) {
@@ -619,29 +629,77 @@ TEST_F(TestMemRowSet, TestScanSnapToExclude) {
// Captures zero rows; a snapshot range [x, x) does not include anything.
for (const auto& s : snaps) {
- NO_FATALS(DumpAndCheck(s, s));
+ NO_FATALS(DumpAndCheck(s, s, boost::none));
}
+ // If we include deleted rows, the row's value will be 1 due to the UPDATE
+ // that preceeded it.
+ boost::optional<int> deleted_v = GetParam() ? boost::optional<int>(1) : boost::none;
+
{
NO_FATALS(DumpAndCheck(snaps[0], snaps[1], 0)); // INSERT
NO_FATALS(DumpAndCheck(snaps[1], snaps[2], 1)); // UPDATE
- NO_FATALS(DumpAndCheck(snaps[2], snaps[3])); // DELETE
+ NO_FATALS(DumpAndCheck(snaps[2], snaps[3], deleted_v)); // DELETE
NO_FATALS(DumpAndCheck(snaps[3], snaps[4], 2)); // REINSERT
}
{
NO_FATALS(DumpAndCheck(snaps[0], snaps[2], 1)); // INSERT, UPDATE
- NO_FATALS(DumpAndCheck(snaps[1], snaps[3])); // UPDATE, DELETE
+ NO_FATALS(DumpAndCheck(snaps[1], snaps[3], deleted_v)); // UPDATE, DELETE
NO_FATALS(DumpAndCheck(snaps[2], snaps[4], 2)); // DELETE, REINSERT
}
{
- NO_FATALS(DumpAndCheck(snaps[0], snaps[3])); // INSERT, UPDATE, DELETE
+ NO_FATALS(DumpAndCheck(snaps[0], snaps[3], deleted_v)); // INSERT, UPDATE, DELETE
NO_FATALS(DumpAndCheck(snaps[1], snaps[4], 2)); // UPDATE, DELETE, REINSERT
}
NO_FATALS(DumpAndCheck(snaps[0], snaps[4], 2)); // INSERT, UPDATE, DELETE, REINSERT
}
+TEST_F(TestMemRowSet, TestScanIncludeDeletedRows) {
+ shared_ptr<MemRowSet> mrs;
+ ASSERT_OK(MemRowSet::Create(0, schema_, log_anchor_registry_.get(),
+ MemTracker::GetRootTracker(), &mrs));
+
+ // row 0 - insert
+ // row 1 - insert, update
+ // row 2 - insert, delete
+ // row 3 - insert, update, delete
+ // row 4 - insert, update, delete, reinsert
+ // row 5 - insert, update, update, delete, reinsert
+ // row 6 - insert, delete, reinsert, delete
+ ASSERT_OK(InsertRow(mrs.get(), "row 0", 0));
+ ASSERT_OK(InsertRow(mrs.get(), "row 1", 0));
+ OperationResultPB result;
+ ASSERT_OK(UpdateRow(mrs.get(), "row 1", 1, &result));
+ ASSERT_OK(InsertRow(mrs.get(), "row 2", 0));
+ ASSERT_OK(DeleteRow(mrs.get(), "row 2", &result));
+ ASSERT_OK(InsertRow(mrs.get(), "row 3", 0));
+ ASSERT_OK(UpdateRow(mrs.get(), "row 3", 1, &result));
+ ASSERT_OK(DeleteRow(mrs.get(), "row 3", &result));
+ ASSERT_OK(InsertRow(mrs.get(), "row 4", 0));
+ ASSERT_OK(UpdateRow(mrs.get(), "row 4", 1, &result));
+ ASSERT_OK(DeleteRow(mrs.get(), "row 4", &result));
+ ASSERT_OK(InsertRow(mrs.get(), "row 4", 2));
+ ASSERT_OK(InsertRow(mrs.get(), "row 5", 0));
+ ASSERT_OK(UpdateRow(mrs.get(), "row 5", 1, &result));
+ ASSERT_OK(UpdateRow(mrs.get(), "row 5", 2, &result));
+ ASSERT_OK(DeleteRow(mrs.get(), "row 5", &result));
+ ASSERT_OK(InsertRow(mrs.get(), "row 5", 3));
+ ASSERT_OK(InsertRow(mrs.get(), "row 6", 0));
+ ASSERT_OK(DeleteRow(mrs.get(), "row 6", &result));
+ ASSERT_OK(InsertRow(mrs.get(), "row 6", 1));
+ ASSERT_OK(DeleteRow(mrs.get(), "row 6", &result));
+
+ RowIteratorOptions opts;
+ opts.projection = &schema_;
+ opts.snap_to_include = MvccSnapshot::CreateSnapshotIncludingAllTransactions();
+ ASSERT_EQ(4, ScanAndCount(mrs.get(), opts));
+
+ opts.include_deleted_rows = true;
+ ASSERT_EQ(7, ScanAndCount(mrs.get(), opts));
+}
+
} // namespace tablet
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/0418b225/src/kudu/tablet/memrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc
index 213b7b9..ec435c5 100644
--- a/src/kudu/tablet/memrowset.cc
+++ b/src/kudu/tablet/memrowset.cc
@@ -526,7 +526,7 @@ Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) {
ApplyStatus apply_status;
RETURN_NOT_OK(ApplyMutationsToProjectedRow(
redo_head, &dst_row, dst->arena(), &apply_status));
- unset_in_sel_vector = apply_status == APPLIED_AND_DELETED ||
+ unset_in_sel_vector = (apply_status == APPLIED_AND_DELETED && !opts_.include_deleted_rows) ||
(apply_status == NONE_APPLIED && insert_excluded);
} else {
// The insertion is too new; the entire row should be omitted.
http://git-wip-us.apache.org/repos/asf/kudu/blob/0418b225/src/kudu/tablet/rowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/rowset.cc b/src/kudu/tablet/rowset.cc
index fb815fd..4e94e6d 100644
--- a/src/kudu/tablet/rowset.cc
+++ b/src/kudu/tablet/rowset.cc
@@ -41,7 +41,8 @@ namespace tablet {
RowIteratorOptions::RowIteratorOptions()
: projection(nullptr),
snap_to_include(MvccSnapshot::CreateSnapshotIncludingAllTransactions()),
- order(OrderMode::UNORDERED) {}
+ order(OrderMode::UNORDERED),
+ include_deleted_rows(false) {}
DuplicatingRowSet::DuplicatingRowSet(RowSetVector old_rowsets,
RowSetVector new_rowsets)
http://git-wip-us.apache.org/repos/asf/kudu/blob/0418b225/src/kudu/tablet/rowset.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/rowset.h b/src/kudu/tablet/rowset.h
index d8e8bec..47ddf82 100644
--- a/src/kudu/tablet/rowset.h
+++ b/src/kudu/tablet/rowset.h
@@ -87,6 +87,11 @@ struct RowIteratorOptions {
//
// Defaults to UNORDERED.
OrderMode order;
+
+ // Whether iteration should include rows whose last mutation was a DELETE.
+ //
+ // Defaults to false.
+ bool include_deleted_rows;
};
class RowSet {