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 2019/06/08 00:23:26 UTC
[kudu] 02/05: KUDU-2809 (2/6): skip unobservable rows when
iterating an MRS
This is an automated email from the ASF dual-hosted git repository.
adar pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 3afbb4af1313a0556c6534c7736746f5a5181342
Author: Adar Dembo <ad...@cloudera.com>
AuthorDate: Sun May 19 21:31:23 2019 -0700
KUDU-2809 (2/6): skip unobservable rows when iterating an MRS
An unobservable row is one whose entire lifespan is contained within the
time range of a diff scan. Such rows should be hidden from the diff scan
results as they represent "changes" that the client wouldn't expect to see.
This patch adds such a heuristic to the MRS iterator. It's relatively
straight-forward because the MRS only stores REDOs and always remembers its
rows' insertion timestamps. Delta stores are more complicated; a patch for
handling unobservable rows for them will follow.
Change-Id: I801991260749d1f810540cb32ec84c3ea6a02160
Reviewed-on: http://gerrit.cloudera.org:8080/13534
Reviewed-by: Mike Percy <mp...@apache.org>
Tested-by: Kudu Jenkins
---
src/kudu/tablet/memrowset-test.cc | 2 +-
src/kudu/tablet/memrowset.cc | 49 ++++++++++++++++++-----
src/kudu/tablet/memrowset.h | 9 ++++-
src/kudu/tablet/tablet-test.cc | 83 +++++++++++++++++++++++++++++++++++++++
4 files changed, 132 insertions(+), 11 deletions(-)
diff --git a/src/kudu/tablet/memrowset-test.cc b/src/kudu/tablet/memrowset-test.cc
index f32002f..0e3b363 100644
--- a/src/kudu/tablet/memrowset-test.cc
+++ b/src/kudu/tablet/memrowset-test.cc
@@ -727,7 +727,7 @@ TEST_P(ParameterizedTestMemRowSet, TestScanSnapToExclude) {
}
{
- NO_FATALS(DumpAndCheck(snaps[0], snaps[3], deleted_v, true)); // INSERT, UPDATE, DELETE
+ NO_FATALS(DumpAndCheck(snaps[0], snaps[3], boost::none)); // INSERT, UPDATE, DELETE
NO_FATALS(DumpAndCheck(snaps[1], snaps[4], 2)); // UPDATE, DELETE, REINSERT
}
diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc
index 9f56851..58cabab 100644
--- a/src/kudu/tablet/memrowset.cc
+++ b/src/kudu/tablet/memrowset.cc
@@ -531,9 +531,10 @@ Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) {
Mutation* redo_head = reinterpret_cast<Mutation*>(
base::subtle::Acquire_Load(reinterpret_cast<AtomicWord*>(&row.header_->redo_head)));
RETURN_NOT_OK(ApplyMutationsToProjectedRow(
- redo_head, &dst_row, dst->arena(), &apply_status));
+ redo_head, &dst_row, dst->arena(), insert_excluded, &apply_status));
unset_in_sel_vector = (apply_status == APPLIED_AND_DELETED && !opts_.include_deleted_rows) ||
- (apply_status == NONE_APPLIED && insert_excluded);
+ (apply_status == NONE_APPLIED && insert_excluded) ||
+ (apply_status == APPLIED_AND_UNOBSERVABLE);
} else {
// The insertion is too new; the entire row should be omitted.
unset_in_sel_vector = true;
@@ -563,7 +564,7 @@ Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) {
Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
const Mutation* mutation_head, RowBlockRow* dst_row, Arena* dst_arena,
- ApplyStatus* apply_status) {
+ bool insert_excluded, ApplyStatus* apply_status) {
ApplyStatus local_apply_status = NONE_APPLIED;
// Fast short-circuit the likely case of a row which was inserted and never
@@ -573,7 +574,23 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
return Status::OK();
}
- bool is_deleted = false;
+ // In order to find unobservable rows, we need to track row liveness at the
+ // start and end of the time range. If a row was dead at both ends, its
+ // lifespan must have been a subset of the time range and it should be
+ // excluded from the results.
+ //
+ // Finding 'is_deleted_end' is relatively straight-forward: we use each
+ // relevant mutation to drive a liveness state machine, and after we're done
+ // applying, 'is_deleted_end' is just the final value of that state machine.
+ //
+ // Finding 'is_deleted_start' is trickier. If the insertion was inside the
+ // time range, we know the value is true because the row was dead prior to the
+ // insertion and the insertion happened after the start of the time range.
+ // However, if the insertion was excluded from the time range, the value is
+ // going to be whatever the value of the liveness state machine was at the
+ // start of the time range.
+ bool is_deleted_start = !insert_excluded;
+ bool is_deleted_end = false;
for (const Mutation *mut = mutation_head;
mut != nullptr;
@@ -591,6 +608,12 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
// count towards the overall "application status".
if (!opts_.snap_to_exclude ||
!opts_.snap_to_exclude->IsCommitted(mut->timestamp_)) {
+
+ // This is the first mutation within the time range, so we may use it to
+ // initialize 'is_deleted_start'.
+ if (local_apply_status == NONE_APPLIED && insert_excluded) {
+ is_deleted_start = is_deleted_end;
+ }
local_apply_status = APPLIED_AND_PRESENT;
}
@@ -600,11 +623,11 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
RowChangeListDecoder decoder(mut->changelist());
RETURN_NOT_OK(decoder.Init());
if (decoder.is_delete()) {
- decoder.TwiddleDeleteStatus(&is_deleted);
+ decoder.TwiddleDeleteStatus(&is_deleted_end);
} else {
DCHECK(decoder.is_update() || decoder.is_reinsert());
if (decoder.is_reinsert()) {
- decoder.TwiddleDeleteStatus(&is_deleted);
+ decoder.TwiddleDeleteStatus(&is_deleted_end);
}
// TODO(todd): this is slow, since it makes multiple passes through the rowchangelist.
@@ -620,9 +643,17 @@ 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 && local_apply_status == APPLIED_AND_PRESENT) {
+ if (opts_.snap_to_exclude && is_deleted_start && is_deleted_end) {
+ // The row's lifespan was a subset of the time range. It can't be observed,
+ // so it should definitely not show up in the results.
+ //
+ // Note: we condition on 'snap_to_exclude' because although insert_excluded
+ // is false on some closed time range scans, it's also false in all open
+ // time range scans, and we don't want this heuristic to fire in the latter case.
+ local_apply_status = APPLIED_AND_UNOBSERVABLE;
+ } else if (is_deleted_end && local_apply_status == APPLIED_AND_PRESENT) {
+ // The row was selected and deleted. It may be omitted from the results,
+ // depending on whether the results should include deleted rows or not.
local_apply_status = APPLIED_AND_DELETED;
}
diff --git a/src/kudu/tablet/memrowset.h b/src/kudu/tablet/memrowset.h
index 752140d..89a02ac 100644
--- a/src/kudu/tablet/memrowset.h
+++ b/src/kudu/tablet/memrowset.h
@@ -566,7 +566,8 @@ class MemRowSet::Iterator : public RowwiseIterator {
Status FetchRows(RowBlock* dst, size_t* fetched);
// Walks the mutations in 'mutation_head', applying relevant ones to 'dst_row'
- // (performing any allocations out of 'dst_arena').
+ // (performing any allocations out of 'dst_arena'). 'insert_excluded' is true
+ // if the row's original insertion took place outside the iterator's time range.
//
// On success, 'apply_status' summarizes the application process.
enum ApplyStatus {
@@ -580,10 +581,16 @@ class MemRowSet::Iterator : public RowwiseIterator {
// 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,
+
+ // Some mutations were applied, but the sequence of applied mutations was
+ // such that clients should never see this row in their output (i.e. the row
+ // was inserted and deleted in the same timestamp).
+ APPLIED_AND_UNOBSERVABLE,
};
Status ApplyMutationsToProjectedRow(const Mutation* mutation_head,
RowBlockRow* dst_row,
Arena* dst_arena,
+ bool insert_excluded,
ApplyStatus* apply_status);
const std::shared_ptr<const MemRowSet> memrowset_;
diff --git a/src/kudu/tablet/tablet-test.cc b/src/kudu/tablet/tablet-test.cc
index 3f46522..d714fad 100644
--- a/src/kudu/tablet/tablet-test.cc
+++ b/src/kudu/tablet/tablet-test.cc
@@ -40,6 +40,7 @@
#include "kudu/common/rowblock.h"
#include "kudu/common/schema.h"
#include "kudu/common/timestamp.h"
+#include "kudu/common/wire_protocol.pb.h"
#include "kudu/fs/block_id.h"
#include "kudu/fs/block_manager.h"
#include "kudu/gutil/gscoped_ptr.h"
@@ -1429,5 +1430,87 @@ TEST_F(TestTabletStringKey, TestSplitKeyRangeWithMinimumValueRowSet) {
}
}
+TYPED_TEST(TestTablet, TestDiffScanUnobservableOperations) {
+ LocalTabletWriter writer(this->tablet().get(), &this->client_schema());
+ vector<LocalTabletWriter::Op> ops;
+
+ // Row 0: INSERT -> DELETE.
+
+ KuduPartialRow insert_row0(&this->client_schema());
+ this->setup_.BuildRow(&insert_row0, 0);
+ ops.emplace_back(RowOperationsPB::INSERT, &insert_row0);
+
+ KuduPartialRow delete_row0(&this->client_schema());
+ this->setup_.BuildRowKey(&delete_row0, 0);
+ ops.emplace_back(RowOperationsPB::DELETE, &delete_row0);
+
+ // Row 1: INSERT -> UPDATE -> DELETE.
+
+ KuduPartialRow insert_row1(&this->client_schema());
+ this->setup_.BuildRow(&insert_row1, 1);
+ ops.emplace_back(RowOperationsPB::INSERT, &insert_row1);
+
+ KuduPartialRow update_row1(&this->client_schema());
+ this->setup_.BuildRowKey(&update_row1, 1);
+ int col_idx = this->client_schema().num_key_columns() == 1 ? 2 : 3;
+ ASSERT_OK(update_row1.SetInt32(col_idx, 10));
+ ops.emplace_back(RowOperationsPB::UPDATE, &update_row1);
+
+ KuduPartialRow delete_row1(&this->client_schema());
+ this->setup_.BuildRowKey(&delete_row1, 1);
+ ops.emplace_back(RowOperationsPB::DELETE, &delete_row1);
+
+ // Row 2: INSERT -> DELETE -> REINSERT -> DELETE.
+
+ KuduPartialRow insert_row2(&this->client_schema());
+ this->setup_.BuildRow(&insert_row2, 2);
+ ops.emplace_back(RowOperationsPB::INSERT, &insert_row2);
+
+ KuduPartialRow first_delete_row2(&this->client_schema());
+ this->setup_.BuildRowKey(&first_delete_row2, 2);
+ ops.emplace_back(RowOperationsPB::DELETE, &first_delete_row2);
+
+ KuduPartialRow reinsert_row2(&this->client_schema());
+ this->setup_.BuildRow(&reinsert_row2, 2);
+ ops.emplace_back(RowOperationsPB::INSERT, &reinsert_row2);
+
+ KuduPartialRow second_delete_row2(&this->client_schema());
+ this->setup_.BuildRowKey(&second_delete_row2, 2);
+ ops.emplace_back(RowOperationsPB::DELETE, &second_delete_row2);
+
+ // Write all operations to the tablet as part of the same batch. This means
+ // that they will all be assigned the same timestamp.
+ ASSERT_OK(writer.WriteBatch(ops));
+
+ // Performs a diff scan, expecting an empty result set because all three rows
+ // are deleted at all times.
+ auto diff_scan_no_rows = [&]() {
+ // Create a projection with an IS_DELETED virtual column.
+ vector<ColumnSchema> col_schemas(this->client_schema().columns());
+ bool read_default = false;
+ col_schemas.emplace_back("is_deleted", IS_DELETED, /*is_nullable=*/ false,
+ &read_default);
+ Schema projection(col_schemas, this->client_schema().num_key_columns());
+
+ // Do the diff scan.
+ RowIteratorOptions opts;
+ opts.projection = &projection;
+ opts.snap_to_exclude = MvccSnapshot(Timestamp(1));
+ opts.snap_to_include = MvccSnapshot(Timestamp(2));
+ opts.include_deleted_rows = true;
+ unique_ptr<RowwiseIterator> iter;
+ ASSERT_OK(this->tablet()->NewRowIterator(std::move(opts), &iter));
+ ASSERT_OK(iter->Init(nullptr));
+ vector<string> lines;
+ ASSERT_OK(IterateToStringList(iter.get(), &lines));
+
+ // Test the results.
+ SCOPED_TRACE(JoinStrings(lines, "\n"));
+ ASSERT_TRUE(lines.empty());
+ };
+
+ NO_FATALS(diff_scan_no_rows());
+}
+
} // namespace tablet
} // namespace kudu