You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2018/12/07 03:47:50 UTC
[1/5] kudu git commit: [test] Break up test output per test case.
Repository: kudu
Updated Branches:
refs/heads/master 94b066bb3 -> ae5eeedba
[test] Break up test output per test case.
Currently, when generating the JUnit XML for each
test suite Gradle appends all of the test output to
stderr and stdout at the suite level. This
change ensures Gradle will break up the output
by each test case instead.
Change-Id: If7564e48cddc7bf7ea7595d925f605802259dced
Reviewed-on: http://gerrit.cloudera.org:8080/12031
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Will Berkeley <wd...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/0008990b
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/0008990b
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/0008990b
Branch: refs/heads/master
Commit: 0008990b6591d8859c5a3ef60f120a5afc7c4406
Parents: 94b066b
Author: Grant Henke <gr...@apache.org>
Authored: Tue Dec 4 14:29:51 2018 -0600
Committer: Grant Henke <gr...@apache.org>
Committed: Thu Dec 6 19:07:50 2018 +0000
----------------------------------------------------------------------
java/gradle/tests.gradle | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/0008990b/java/gradle/tests.gradle
----------------------------------------------------------------------
diff --git a/java/gradle/tests.gradle b/java/gradle/tests.gradle
index c5110fb..9c51858 100644
--- a/java/gradle/tests.gradle
+++ b/java/gradle/tests.gradle
@@ -29,6 +29,9 @@ tasks.withType(Test) {
exceptionFormat = "full"
}
+ // Break up the stdout and stderr logging by test case.
+ reports.junitXml.outputPerTestCase = true
+
// Fork the jvm for every test class.
// This takes more time and resources but ensures isolation.
forkEvery 1
[5/5] kudu git commit: (05/05) delta_applier: support for diff scan
style iteration
Posted by mp...@apache.org.
(05/05) delta_applier: support for diff scan style iteration
This patch incorporates all of the previous work to provide diff scan
support for an entire DiskRowSet. There's a new "fuzzy" test in
diskrowset-test that exercises this functionality.
Change-Id: I895350a3c7728df88a87a36b904ffdea4f3f6b7a
Reviewed-on: http://gerrit.cloudera.org:8080/11860
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/ae5eeedb
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/ae5eeedb
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/ae5eeedb
Branch: refs/heads/master
Commit: ae5eeedba49bc1db86e5f63f1e642cc0068dcf81
Parents: c931e57
Author: Adar Dembo <ad...@cloudera.com>
Authored: Thu Nov 1 21:04:55 2018 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Fri Dec 7 03:47:32 2018 +0000
----------------------------------------------------------------------
src/kudu/tablet/delta_applier.cc | 36 ++++-
src/kudu/tablet/delta_applier.h | 6 +-
src/kudu/tablet/delta_tracker.cc | 2 +-
src/kudu/tablet/delta_tracker.h | 2 +-
src/kudu/tablet/diskrowset-test.cc | 257 ++++++++++++++++++++++++++++++++
5 files changed, 294 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/ae5eeedb/src/kudu/tablet/delta_applier.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_applier.cc b/src/kudu/tablet/delta_applier.cc
index 459b5a9..7c767a6 100644
--- a/src/kudu/tablet/delta_applier.cc
+++ b/src/kudu/tablet/delta_applier.cc
@@ -22,10 +22,13 @@
#include <utility>
#include <vector>
+#include <boost/optional/optional.hpp>
#include <glog/logging.h>
#include "kudu/common/column_materialization_context.h"
+#include "kudu/common/rowblock.h"
#include "kudu/tablet/delta_store.h"
+#include "kudu/tablet/rowset.h"
#include "kudu/util/status.h"
using std::shared_ptr;
@@ -36,15 +39,16 @@ namespace kudu {
class ScanSpec;
class Schema;
-class SelectionVector;
struct IteratorStats;
namespace tablet {
// Construct. The base_iter and delta_iter should not be Initted.
-DeltaApplier::DeltaApplier(shared_ptr<CFileSet::Iterator> base_iter,
+DeltaApplier::DeltaApplier(RowIteratorOptions opts,
+ shared_ptr<CFileSet::Iterator> base_iter,
unique_ptr<DeltaIterator> delta_iter)
- : base_iter_(std::move(base_iter)),
+ : opts_(std::move(opts)),
+ base_iter_(std::move(base_iter)),
delta_iter_(std::move(delta_iter)),
first_prepare_(true) {}
@@ -89,7 +93,12 @@ Status DeltaApplier::PrepareBatch(size_t *nrows) {
first_prepare_ = false;
}
RETURN_NOT_OK(base_iter_->PrepareBatch(nrows));
- RETURN_NOT_OK(delta_iter_->PrepareBatch(*nrows, DeltaIterator::PREPARE_FOR_APPLY));
+ int prepare_flags = DeltaIterator::PREPARE_FOR_APPLY;
+ if (opts_.snap_to_exclude) {
+ // See InitializeSelectionVector() below.
+ prepare_flags |= DeltaIterator::PREPARE_FOR_SELECT;
+ }
+ RETURN_NOT_OK(delta_iter_->PrepareBatch(*nrows, prepare_flags));
return Status::OK();
}
@@ -99,8 +108,23 @@ Status DeltaApplier::FinishBatch() {
Status DeltaApplier::InitializeSelectionVector(SelectionVector *sel_vec) {
DCHECK(!first_prepare_) << "PrepareBatch() must be called at least once";
- RETURN_NOT_OK(base_iter_->InitializeSelectionVector(sel_vec));
- return delta_iter_->ApplyDeletes(sel_vec);
+
+ // A diff scan will set both 'snap_to_exclude' and 'include_deleted_rows'.
+ // The result: it'll initialize the selection vector using any delta that
+ // meets the select criteria rather than just using a DELETE that meets the
+ // apply criteria.
+ //
+ // See delta_relevancy.h for more details.
+ if (opts_.snap_to_exclude) {
+ sel_vec->SetAllFalse();
+ RETURN_NOT_OK(delta_iter_->SelectUpdates(sel_vec));
+ } else {
+ RETURN_NOT_OK(base_iter_->InitializeSelectionVector(sel_vec));
+ }
+ if (!opts_.include_deleted_rows) {
+ RETURN_NOT_OK(delta_iter_->ApplyDeletes(sel_vec));
+ }
+ return Status::OK();
}
Status DeltaApplier::MaterializeColumn(ColumnMaterializationContext *ctx) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/ae5eeedb/src/kudu/tablet/delta_applier.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_applier.h b/src/kudu/tablet/delta_applier.h
index 00e37ef..cdd99aa 100644
--- a/src/kudu/tablet/delta_applier.h
+++ b/src/kudu/tablet/delta_applier.h
@@ -28,6 +28,7 @@
#include "kudu/gutil/macros.h"
#include "kudu/gutil/port.h"
#include "kudu/tablet/cfile_set.h"
+#include "kudu/tablet/rowset.h"
#include "kudu/util/status.h"
namespace kudu {
@@ -78,10 +79,13 @@ class DeltaApplier : public ColumnwiseIterator {
DISALLOW_COPY_AND_ASSIGN(DeltaApplier);
// Construct. The base_iter and delta_iter should not be Initted.
- DeltaApplier(std::shared_ptr<CFileSet::Iterator> base_iter,
+ DeltaApplier(RowIteratorOptions opts,
+ std::shared_ptr<CFileSet::Iterator> base_iter,
std::unique_ptr<DeltaIterator> delta_iter);
virtual ~DeltaApplier();
+ const RowIteratorOptions opts_;
+
std::shared_ptr<CFileSet::Iterator> base_iter_;
std::unique_ptr<DeltaIterator> delta_iter_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/ae5eeedb/src/kudu/tablet/delta_tracker.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_tracker.cc b/src/kudu/tablet/delta_tracker.cc
index b10465b..8c3df30 100644
--- a/src/kudu/tablet/delta_tracker.cc
+++ b/src/kudu/tablet/delta_tracker.cc
@@ -607,7 +607,7 @@ Status DeltaTracker::WrapIterator(const shared_ptr<CFileSet::Iterator> &base,
unique_ptr<DeltaIterator> iter;
RETURN_NOT_OK(NewDeltaIterator(opts, &iter));
- out->reset(new DeltaApplier(base, std::move(iter)));
+ out->reset(new DeltaApplier(opts, base, std::move(iter)));
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/ae5eeedb/src/kudu/tablet/delta_tracker.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_tracker.h b/src/kudu/tablet/delta_tracker.h
index 7a1e76e..f1bf87a 100644
--- a/src/kudu/tablet/delta_tracker.h
+++ b/src/kudu/tablet/delta_tracker.h
@@ -52,8 +52,8 @@ class OpId;
}
namespace fs {
-struct IOContext;
class WritableBlock;
+struct IOContext;
}
namespace log {
http://git-wip-us.apache.org/repos/asf/kudu/blob/ae5eeedb/src/kudu/tablet/diskrowset-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/diskrowset-test.cc b/src/kudu/tablet/diskrowset-test.cc
index 7e89122..3889162 100644
--- a/src/kudu/tablet/diskrowset-test.cc
+++ b/src/kudu/tablet/diskrowset-test.cc
@@ -18,6 +18,7 @@
#include "kudu/tablet/diskrowset.h"
#include <algorithm>
+#include <cstddef>
#include <cstdint>
#include <memory>
#include <ostream>
@@ -26,22 +27,32 @@
#include <unordered_set>
#include <vector>
+#include <boost/optional/optional.hpp>
#include <gflags/gflags.h>
#include <gflags/gflags_declare.h>
#include <glog/logging.h>
+#include <glog/stl_logging.h>
#include <gtest/gtest.h>
#include "kudu/clock/clock.h"
+#include "kudu/clock/logical_clock.h"
+#include "kudu/common/common.pb.h"
#include "kudu/common/iterator.h"
#include "kudu/common/row.h"
#include "kudu/common/row_changelist.h"
+#include "kudu/common/rowid.h"
#include "kudu/common/schema.h"
#include "kudu/common/timestamp.h"
+#include "kudu/consensus/log_anchor_registry.h"
+#include "kudu/consensus/opid.pb.h"
+#include "kudu/consensus/opid_util.h"
#include "kudu/fs/block_id.h"
+#include "kudu/fs/io_context.h"
#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/stringpiece.h"
+#include "kudu/gutil/strings/substitute.h"
#include "kudu/tablet/compaction.h"
#include "kudu/tablet/delta_key.h"
#include "kudu/tablet/delta_store.h"
@@ -54,12 +65,15 @@
#include "kudu/tablet/tablet-test-util.h"
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet.pb.h"
+#include "kudu/tablet/tablet_mem_trackers.h"
#include "kudu/util/bloom_filter.h"
#include "kudu/util/faststring.h"
+#include "kudu/util/random.h"
#include "kudu/util/slice.h"
#include "kudu/util/status.h"
#include "kudu/util/stopwatch.h"
#include "kudu/util/test_macros.h"
+#include "kudu/util/test_util.h"
DEFINE_double(update_fraction, 0.1f, "fraction of rows to update");
DECLARE_bool(cfile_lazy_open);
@@ -70,11 +84,14 @@ DECLARE_double(tablet_delta_store_major_compact_min_ratio);
DECLARE_int32(tablet_delta_store_minor_compact_max);
using std::is_sorted;
+using std::make_tuple;
using std::shared_ptr;
using std::string;
using std::unique_ptr;
using std::unordered_set;
+using std::tuple;
using std::vector;
+using strings::Substitute;
namespace kudu {
namespace tablet {
@@ -678,5 +695,245 @@ TEST_F(TestRowSet, TestDiskSizeEstimation) {
ASSERT_GT(drss.redo_deltas_size, 0);
}
+class DiffScanRowSetTest : public KuduRowSetTest,
+ public ::testing::WithParamInterface<tuple<bool, bool>> {
+ public:
+ DiffScanRowSetTest()
+ : KuduRowSetTest(CreateTestSchema()),
+ op_id_(consensus::MaximumOpId()),
+ clock_(clock::LogicalClock::CreateStartingAt(Timestamp::kInitialTimestamp)) {
+ }
+
+ protected:
+ static Schema CreateTestSchema() {
+ SchemaBuilder builder;
+ CHECK_OK(builder.AddKeyColumn("key", UINT32));
+ CHECK_OK(builder.AddColumn("val1", UINT32));
+ CHECK_OK(builder.AddColumn("val2", UINT32));
+ return builder.BuildWithoutIds();
+ }
+
+ consensus::OpId op_id_;
+ scoped_refptr<clock::Clock> clock_;
+ MvccManager mvcc_;
+};
+
+// Tests the Cartesian product of two boolean parameters:
+// 1. Whether to include deleted rows in the scan.
+// 2. Whether to include the "is deleted" virtual column in the scan's projection.
+INSTANTIATE_TEST_CASE_P(RowIteratorOptionsPermutations, DiffScanRowSetTest,
+ ::testing::Combine(::testing::Bool(),
+ ::testing::Bool()));
+
+// Tests diff scans on a diskrowset. The rowset is generated with only a handful
+// of rows, but we randomly flush/compact between each update operation so that
+// the test operates on a variety of different on-disk and in-memory layouts.
+TEST_P(DiffScanRowSetTest, TestFuzz) {
+ fs::IOContext test_context;
+
+ // Create and open a DRS with four rows.
+ shared_ptr<DiskRowSet> rs;
+ {
+ DiskRowSetWriter drsw(rowset_meta_.get(), &schema_,
+ BloomFilterSizing::BySizeAndFPRate(32 * 1024, 0.01f));
+ ASSERT_OK(drsw.Open());
+
+ RowBuilder rb(schema_);
+ for (int i = 0; i < 4; i++) {
+ rb.Reset();
+ rb.AddUint32(i);
+ rb.AddUint32(i * 2);
+ rb.AddUint32(i * 3);
+ ASSERT_OK(WriteRow(rb.data(), &drsw));
+ }
+ ASSERT_OK(drsw.Finish());
+ ASSERT_OK(DiskRowSet::Open(rowset_meta_, new log::LogAnchorRegistry(),
+ TabletMemTrackers(), &test_context, &rs));
+ }
+
+ // Run the diff scan test. Scan time boundaries are given by 'ts1_val' and
+ // 'ts2_val'. The expected results are given by 'expected_rows'.
+ using RowTuple = tuple<uint32_t, uint32_t, uint32_t, bool>;
+ auto run_test = [&](uint64_t ts1_val,
+ uint64_t ts2_val,
+ vector<RowTuple> expected_rows) {
+ bool include_deleted_rows = std::get<0>(GetParam());
+ bool add_vc_is_deleted = std::get<1>(GetParam());
+
+ // Create a projection of the schema, adding the IS_DELETED virtual
+ // column if desired.
+ vector<ColumnSchema> col_schemas;
+ vector<ColumnId> col_ids;
+ for (int i = 0; i < schema_.num_columns(); i++) {
+ col_schemas.push_back(schema_.column(i));
+ col_ids.push_back(schema_.column_id(i));
+ }
+ if (add_vc_is_deleted) {
+ bool read_default = false;
+ col_schemas.emplace_back("is_deleted", IS_DELETED, /*is_nullable=*/ false,
+ &read_default);
+ col_ids.emplace_back(schema_.max_col_id() + 1);
+ }
+ Schema projection(col_schemas, col_ids, 1);
+
+ // Set up the iterator.
+ Timestamp ts1(ts1_val);
+ Timestamp ts2(ts2_val);
+ SCOPED_TRACE(Substitute("Scanning at $0,$1 with schema $2",
+ ts1.ToString(), ts2.ToString(), projection.ToString()));
+ RowIteratorOptions opts;
+ opts.include_deleted_rows = include_deleted_rows;
+ opts.projection = &projection;
+ opts.snap_to_exclude = MvccSnapshot(ts1);
+ opts.snap_to_include = MvccSnapshot(ts2);
+ gscoped_ptr<RowwiseIterator> iter;
+ ASSERT_OK(rs->NewRowIterator(opts, &iter));
+ ASSERT_OK(iter->Init(nullptr));
+
+ // Scan the data out of the iterator and test it.
+ vector<string> lines;
+ ASSERT_OK(IterateToStringList(iter.get(), &lines));
+ if (!include_deleted_rows) {
+ vector<RowTuple> without_deleted_rows;
+ for (const auto& e : expected_rows) {
+ if (!std::get<3>(e)) {
+ without_deleted_rows.emplace_back(e);
+ }
+ }
+ expected_rows = without_deleted_rows;
+ }
+ ASSERT_EQ(expected_rows.size(), lines.size()) << lines;
+ for (int i = 0; i < expected_rows.size(); i++) {
+ string expected_is_deleted = add_vc_is_deleted ? Substitute(
+ ", is_deleted is_deleted=$0", std::get<3>(expected_rows[i])) : "";
+ string expected_line = Substitute("(uint32 key=$0, "
+ "uint32 val1=$1, "
+ "uint32 val2=$2$3)",
+ std::get<0>(expected_rows[i]),
+ std::get<1>(expected_rows[i]),
+ std::get<2>(expected_rows[i]),
+ expected_is_deleted);
+ ASSERT_EQ(expected_line, lines[i]);
+ }
+ };
+
+ // Mutate a particular column in the diskrowset given by 'row_idx' and
+ // 'col_idx'. If 'val' is unset, deletes the row.
+ auto mutate_row = [&](rowid_t row_idx,
+ size_t col_idx,
+ boost::optional<uint32_t> val) {
+ // Build the mutation.
+ faststring buf;
+ RowChangeListEncoder enc(&buf);
+ if (val) {
+ enc.AddColumnUpdate(schema_.column(col_idx),
+ schema_.column_id(col_idx),
+ val.get_ptr());
+ } else {
+ enc.SetToDelete();
+ }
+
+ // Build the row key.
+ RowBuilder rb(schema_.CreateKeyProjection());
+ rb.AddUint32(row_idx);
+ RowSetKeyProbe probe(rb.row());
+
+ // Apply the mutation.
+ ScopedTransaction tx(&mvcc_, clock_->Now());
+ tx.StartApplying();
+ ProbeStats stats;
+ OperationResultPB result;
+ ASSERT_OK(rs->MutateRow(tx.timestamp(), probe, enc.as_changelist(), op_id_,
+ &test_context, &stats, &result));
+ tx.Commit();
+ };
+
+ Random prng(SeedRandom());
+
+ // Possibly apply a randomly chosen delta flush or compaction operation.
+ auto maybe_flush_compact = [&]() {
+ // 25% for a minor delta compaction.
+ // 25% for a major delta compaction.
+ // 50% for a DMS flush.
+ int r = prng.Uniform(4);
+ if (r == 0) {
+ ASSERT_OK(rs->MinorCompactDeltaStores(&test_context));
+ } else if (r == 1) {
+ ASSERT_OK(rs->MajorCompactDeltaStores(&test_context, HistoryGcOpts::Disabled()));
+ } else {
+ ASSERT_OK(rs->FlushDeltas(&test_context));
+ }
+ };
+
+ // Update the rows in the diskrowset.
+ NO_FATALS(maybe_flush_compact());
+ NO_FATALS(mutate_row(1, 2, 1000)); // ts 1
+ NO_FATALS(maybe_flush_compact());
+ NO_FATALS(mutate_row(1, 1, 200)); // ts 2
+ NO_FATALS(maybe_flush_compact());
+ NO_FATALS(mutate_row(2, 1, 300)); // ts 3
+ NO_FATALS(maybe_flush_compact());
+ NO_FATALS(mutate_row(3, 1, 400)); // ts 4
+ NO_FATALS(maybe_flush_compact());
+ NO_FATALS(mutate_row(3, 1, boost::none)); // ts 5
+ NO_FATALS(maybe_flush_compact());
+
+ // Run the diff scan test on every permutation of time bounds.
+ //
+ // Note: a regular DRS would include an UNDO DELETE for every INSERT by virtue
+ // of having been flushed from an MRS. That's not the case here, and as a
+ // result, the insertions themselves aren't associated with any timestamps and
+ // thus are not captured by these diff scans.
+
+ // The time bounds specify an empty range so nothing is captured.
+ for (int i = 0; i < 7; i++) {
+ NO_FATALS(run_test(i, i, {}));
+ }
+
+ NO_FATALS(run_test(0, 1, {}));
+ NO_FATALS(run_test(1, 2, { make_tuple(1, 2, 1000, false) }));
+ NO_FATALS(run_test(2, 3, { make_tuple(1, 200, 1000, false) }));
+ NO_FATALS(run_test(3, 4, { make_tuple(2, 300, 6, false) }));
+ NO_FATALS(run_test(4, 5, { make_tuple(3, 400, 9, false) }));
+ NO_FATALS(run_test(5, 6, { make_tuple(3, 400, 9, true) }));
+
+ NO_FATALS(run_test(0, 2, { make_tuple(1, 2, 1000, false) }));
+ NO_FATALS(run_test(1, 3, { make_tuple(1, 200, 1000, false) }));
+ NO_FATALS(run_test(2, 4, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false) }));
+ NO_FATALS(run_test(3, 5, { make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, false) }));
+ NO_FATALS(run_test(4, 6, { make_tuple(3, 400, 9, true) }));
+
+ NO_FATALS(run_test(0, 3, { make_tuple(1, 200, 1000, false) }));
+ NO_FATALS(run_test(1, 4, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false) }));
+ NO_FATALS(run_test(2, 5, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, false) }));
+ NO_FATALS(run_test(3, 6, { make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, true) }));
+
+ NO_FATALS(run_test(0, 4, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false) }));
+ NO_FATALS(run_test(1, 5, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, false) }));
+ NO_FATALS(run_test(2, 6, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, true) }));
+
+ NO_FATALS(run_test(0, 5, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, false) }));
+ NO_FATALS(run_test(1, 6, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, true) }));
+
+ NO_FATALS(run_test(0, 6, { make_tuple(1, 200, 1000, false),
+ make_tuple(2, 300, 6, false),
+ make_tuple(3, 400, 9, true) }));
+}
+
} // namespace tablet
} // namespace kudu
[2/5] kudu git commit: (02/05) delta_store: convert
DeltaIterator::PrepareBatch flags into bitfield
Posted by mp...@apache.org.
(02/05) delta_store: convert DeltaIterator::PrepareBatch flags into bitfield
A follow-on patch will introduce DeltaIterator::SelectUpdates(), a new
method for extracting delta information from a DeltaIterator. As it is
only to be used by diff scans and will require dedicated in-memory state,
it'll also come with a new DeltaIterator::PrepareBatch() flag.
However, diff scans will need to use both SelectUpdates() and
ApplyUpdates(), and it'd be a shame to have to reseek and reprepare just to
use both extraction methods. As such, this patch makes it possible to
prepare a batch for use in multiple delta extraction methods.
Change-Id: Id500ec3cc9459a78ae1098be3f3cd0cacb7a7a1a
Reviewed-on: http://gerrit.cloudera.org:8080/11857
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/d927675b
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/d927675b
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/d927675b
Branch: refs/heads/master
Commit: d927675b0af52adfd03a52265337ee1dc1609fa4
Parents: 0008990
Author: Adar Dembo <ad...@cloudera.com>
Authored: Tue Sep 11 16:27:56 2018 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Fri Dec 7 03:43:24 2018 +0000
----------------------------------------------------------------------
src/kudu/tablet/delta_iterator_merger.cc | 4 +-
src/kudu/tablet/delta_iterator_merger.h | 2 +-
src/kudu/tablet/delta_store.cc | 34 ++++++---------
src/kudu/tablet/delta_store.h | 62 ++++++++++++---------------
src/kudu/tablet/deltafile.cc | 4 +-
src/kudu/tablet/deltafile.h | 2 +-
src/kudu/tablet/deltamemstore.cc | 4 +-
src/kudu/tablet/deltamemstore.h | 2 +-
src/kudu/tablet/tablet-test-util.h | 21 +++------
9 files changed, 55 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_iterator_merger.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_iterator_merger.cc b/src/kudu/tablet/delta_iterator_merger.cc
index 23ed670..f6f5840 100644
--- a/src/kudu/tablet/delta_iterator_merger.cc
+++ b/src/kudu/tablet/delta_iterator_merger.cc
@@ -57,9 +57,9 @@ Status DeltaIteratorMerger::SeekToOrdinal(rowid_t idx) {
return Status::OK();
}
-Status DeltaIteratorMerger::PrepareBatch(size_t nrows, PrepareFlag flag) {
+Status DeltaIteratorMerger::PrepareBatch(size_t nrows, int prepare_flags) {
for (const unique_ptr<DeltaIterator> &iter : iters_) {
- RETURN_NOT_OK(iter->PrepareBatch(nrows, flag));
+ RETURN_NOT_OK(iter->PrepareBatch(nrows, prepare_flags));
}
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_iterator_merger.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_iterator_merger.h b/src/kudu/tablet/delta_iterator_merger.h
index e5e3bb1..a8efeb6 100644
--- a/src/kudu/tablet/delta_iterator_merger.h
+++ b/src/kudu/tablet/delta_iterator_merger.h
@@ -60,7 +60,7 @@ class DeltaIteratorMerger : public DeltaIterator {
Status SeekToOrdinal(rowid_t idx) override;
- Status PrepareBatch(size_t nrows, PrepareFlag flag) override;
+ Status PrepareBatch(size_t nrows, int prepare_flags) override;
Status ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) override;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_store.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.cc b/src/kudu/tablet/delta_store.cc
index 13b1e4a..bb7a91f 100644
--- a/src/kudu/tablet/delta_store.cc
+++ b/src/kudu/tablet/delta_store.cc
@@ -107,7 +107,7 @@ DeltaPreparer<Traits>::DeltaPreparer(RowIteratorOptions opts)
: opts_(std::move(opts)),
cur_prepared_idx_(0),
prev_prepared_idx_(0),
- prepared_for_(NOT_PREPARED),
+ prepared_flags_(DeltaIterator::PREPARE_NONE),
deletion_state_(UNKNOWN) {
}
@@ -115,11 +115,13 @@ template<class Traits>
void DeltaPreparer<Traits>::Seek(rowid_t row_idx) {
cur_prepared_idx_ = row_idx;
prev_prepared_idx_ = row_idx;
- prepared_for_ = NOT_PREPARED;
+ prepared_flags_ = DeltaIterator::PREPARE_NONE;
}
template<class Traits>
-void DeltaPreparer<Traits>::Start(DeltaIterator::PrepareFlag flag) {
+void DeltaPreparer<Traits>::Start(int prepare_flags) {
+ DCHECK_NE(prepare_flags, DeltaIterator::PREPARE_NONE);
+ prepared_flags_ = prepare_flags;
if (updates_by_col_.empty()) {
updates_by_col_.resize(opts_.projection->num_columns());
}
@@ -130,16 +132,6 @@ void DeltaPreparer<Traits>::Start(DeltaIterator::PrepareFlag flag) {
reinserted_.clear();
prepared_deltas_.clear();
deletion_state_ = UNKNOWN;
- switch (flag) {
- case DeltaIterator::PREPARE_FOR_APPLY:
- prepared_for_ = PREPARED_FOR_APPLY;
- break;
- case DeltaIterator::PREPARE_FOR_COLLECT:
- prepared_for_ = PREPARED_FOR_COLLECT;
- break;
- default:
- LOG(FATAL) << "Unknown prepare flag: " << flag;
- }
}
template<class Traits>
@@ -157,7 +149,7 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
}
MaybeProcessPreviousRowChange(key.row_idx());
- if (prepared_for_ == PREPARED_FOR_APPLY) {
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY) {
RowChangeListDecoder decoder((RowChangeList(val)));
if (Traits::kInitializeDecodersWithSafetyChecks) {
RETURN_NOT_OK(decoder.Init());
@@ -201,8 +193,8 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
}
}
}
- } else {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_COLLECT);
+ }
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT) {
PreparedDelta d;
d.key = key;
d.val = val;
@@ -216,7 +208,7 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
template<class Traits>
Status DeltaPreparer<Traits>::ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, dst->nrows());
const ColumnSchema* col_schema = &opts_.projection->column(col_to_apply);
@@ -236,7 +228,7 @@ Status DeltaPreparer<Traits>::ApplyUpdates(size_t col_to_apply, ColumnBlock* dst
template<class Traits>
Status DeltaPreparer<Traits>::ApplyDeletes(SelectionVector* sel_vec) {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, sel_vec->nrows());
for (const auto& row_id : deleted_) {
@@ -254,7 +246,7 @@ Status DeltaPreparer<Traits>::ApplyDeletes(SelectionVector* sel_vec) {
template<class Traits>
Status DeltaPreparer<Traits>::CollectMutations(vector<Mutation*>* dst, Arena* arena) {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_COLLECT);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, dst->size());
for (const PreparedDelta& src : prepared_deltas_) {
DeltaKey key = src.key;
@@ -306,7 +298,7 @@ Status DeltaPreparer<Traits>::FilterColumnIdsAndCollectDeltas(
template<class Traits>
bool DeltaPreparer<Traits>::MayHaveDeltas() const {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
if (!deleted_.empty()) {
return true;
}
@@ -323,7 +315,7 @@ bool DeltaPreparer<Traits>::MayHaveDeltas() const {
template<class Traits>
void DeltaPreparer<Traits>::MaybeProcessPreviousRowChange(boost::optional<rowid_t> cur_row_idx) {
- if (prepared_for_ == PREPARED_FOR_APPLY &&
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY &&
last_added_idx_ &&
(!cur_row_idx || cur_row_idx != *last_added_idx_)) {
switch (deletion_state_) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_store.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.h b/src/kudu/tablet/delta_store.h
index 12ec9c6..955f4a7 100644
--- a/src/kudu/tablet/delta_store.h
+++ b/src/kudu/tablet/delta_store.h
@@ -198,22 +198,37 @@ class DeltaIterator : public PreparedDeltas {
// block, and must be called at least once prior to PrepareBatch().
virtual Status SeekToOrdinal(rowid_t idx) = 0;
- // Argument to PrepareBatch(). See below.
- enum PrepareFlag {
- PREPARE_FOR_APPLY,
- PREPARE_FOR_COLLECT
- };
-
// Prepare to apply deltas to a block of rows. This takes a consistent snapshot
// of all updates to the next 'nrows' rows, so that subsequent calls to a
// PreparedDeltas method will not cause any "tearing"/non-atomicity.
//
- // 'flag' denotes whether the batch will be used for collecting mutations or
- // for applying them. Some implementations may choose to prepare differently.
+ // 'prepare_flags' is a bitfield describing what operation(s) the batch will
+ // be used for; some implementations may choose to prepare differently.
+ // PREPARE_NONE is an invalid value; it is only used internally.
//
// Each time this is called, the iterator is advanced by the full length
// of the previously prepared block.
- virtual Status PrepareBatch(size_t nrows, PrepareFlag flag) = 0;
+ enum {
+ // There are no prepared blocks. Attempts to call a PreparedDeltas function
+ // will fail.
+ PREPARE_NONE = 0,
+
+ // Prepare a batch of deltas for applying. All deltas in the batch will be
+ // decoded. Operations affecting row data (i.e. UPDATEs and REINSERTs) will
+ // be coalesced into a column-major data structure suitable for
+ // ApplyUpdates. Operations affecting row lifecycle (i.e. DELETES and
+ // REINSERTs) will be coalesced into a row-major data structure suitable for ApplyDeletes.
+ //
+ // On success, ApplyUpdates and ApplyDeltas will be callable.
+ PREPARE_FOR_APPLY = 1 << 0,
+
+ // Prepare a batch of deltas for collecting. Deltas will remain encoded and
+ // in the order that they were loaded from the backing store.
+ //
+ // On success, CollectMutations and FilterColumnIdsAndCollectDeltas will be callable.
+ PREPARE_FOR_COLLECT = 1 << 1
+ };
+ virtual Status PrepareBatch(size_t nrows, int prepare_flags) = 0;
// Returns true if there are any more rows left in this iterator.
virtual bool HasNext() = 0;
@@ -266,7 +281,7 @@ class DeltaPreparer : public PreparedDeltas {
// on the part of a DeltaIterator.
//
// Call at the beginning of DeltaIterator::PrepareBatch.
- void Start(DeltaIterator::PrepareFlag flag);
+ void Start(int prepare_flags);
// Updates internal state to reflect the end of delta batch preparation on the
// part of a DeltaIterator.
@@ -329,30 +344,9 @@ class DeltaPreparer : public PreparedDeltas {
boost::optional<rowid_t> last_added_idx_;
// Whether there are any prepared blocks.
- enum PreparedFor {
- // There are no prepared blocks. Attempts to call a PreparedDeltas function
- // will fail.
- NOT_PREPARED,
-
- // The DeltaPreparer has prepared a batch of deltas for applying. All deltas
- // in the batch have been decoded. Operations affecting row data (i.e.
- // UPDATEs and REINSERTs) have been coalesced into a column-major data
- // structure suitable for ApplyUpdates. Operations affecting row lifecycle
- // (i.e. DELETES and REINSERTs) have been coalesced into a row-major data
- // structure suitable for ApplyDeletes.
- //
- // ApplyUpdates and ApplyDeltas are now callable.
- PREPARED_FOR_APPLY,
-
- // The DeltaPreparer has prepared a batch of deltas for collecting. Deltas
- // remain encoded and in the order that they were loaded from the backing store.
- //
- // CollectMutations and FilterColumnIdsAndCollectDeltas are now callable.
- PREPARED_FOR_COLLECT
- };
- PreparedFor prepared_for_;
+ int prepared_flags_;
- // State when prepared_for_ == PREPARED_FOR_APPLY
+ // State when prepared_flags_ & PREPARED_FOR_APPLY
// ------------------------------------------------------------
struct ColumnUpdate {
rowid_t row_id;
@@ -372,7 +366,7 @@ class DeltaPreparer : public PreparedDeltas {
};
RowDeletionState deletion_state_;
- // State when prepared_for_ == PREPARED_FOR_COLLECT
+ // State when prepared_flags_ & PREPARED_FOR_COLLECT
// ------------------------------------------------------------
struct PreparedDelta {
DeltaKey key;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltafile.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.cc b/src/kudu/tablet/deltafile.cc
index 575895a..7ca6719 100644
--- a/src/kudu/tablet/deltafile.cc
+++ b/src/kudu/tablet/deltafile.cc
@@ -532,7 +532,7 @@ string DeltaFileIterator<Type>::PreparedDeltaBlock::ToString() const {
}
template<DeltaType Type>
-Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, PrepareFlag flag) {
+Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, int prepare_flags) {
DCHECK(initted_) << "Must call Init()";
DCHECK(exhausted_ || index_iter_) << "Must call SeekToOrdinal()";
@@ -587,7 +587,7 @@ Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, PrepareFlag flag) {
#endif
prepared_ = true;
- preparer_.Start(flag);
+ preparer_.Start(prepare_flags);
RETURN_NOT_OK(AddDeltas(start_row, stop_row));
preparer_.Finish(nrows);
return Status::OK();
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltafile.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.h b/src/kudu/tablet/deltafile.h
index e99a4f4..217dd4c 100644
--- a/src/kudu/tablet/deltafile.h
+++ b/src/kudu/tablet/deltafile.h
@@ -227,7 +227,7 @@ class DeltaFileIterator : public DeltaIterator {
Status SeekToOrdinal(rowid_t idx) override;
- Status PrepareBatch(size_t nrows, PrepareFlag flag) override;
+ Status PrepareBatch(size_t nrows, int prepare_flags) override;
Status ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) override;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltamemstore.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltamemstore.cc b/src/kudu/tablet/deltamemstore.cc
index 8676d33..a25c110 100644
--- a/src/kudu/tablet/deltamemstore.cc
+++ b/src/kudu/tablet/deltamemstore.cc
@@ -217,7 +217,7 @@ Status DMSIterator::SeekToOrdinal(rowid_t row_idx) {
return Status::OK();
}
-Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) {
+Status DMSIterator::PrepareBatch(size_t nrows, int prepare_flags) {
// This current implementation copies the whole batch worth of deltas
// into a buffer local to this iterator, after filtering out deltas which
// aren't yet committed in the current MVCC snapshot. The theory behind
@@ -233,8 +233,8 @@ Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) {
DCHECK(initted_) << "must init";
rowid_t start_row = preparer_.cur_prepared_idx();
rowid_t stop_row = start_row + nrows - 1;
- preparer_.Start(flag);
+ preparer_.Start(prepare_flags);
bool finished_row = false;
while (iter_->IsValid()) {
Slice key_slice, val;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltamemstore.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltamemstore.h b/src/kudu/tablet/deltamemstore.h
index d9f094b..a5a92f7 100644
--- a/src/kudu/tablet/deltamemstore.h
+++ b/src/kudu/tablet/deltamemstore.h
@@ -199,7 +199,7 @@ class DMSIterator : public DeltaIterator {
Status SeekToOrdinal(rowid_t row_idx) override;
- Status PrepareBatch(size_t nrows, PrepareFlag flag) override;
+ Status PrepareBatch(size_t nrows, int prepare_flags) override;
Status ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) override;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/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 89a816e..87daee6 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -809,15 +809,15 @@ void RunDeltaFuzzTest(const DeltaStore& store,
std::unique_ptr<DeltaIterator> iter(raw_iter);
ASSERT_OK(iter->Init(nullptr));
- // Run PREPARE_FOR_APPLY method tests in batches, in case there's some bug
- // related to batching.
+ // Run tests in batches, in case there's some bug related to batching.
ASSERT_OK(iter->SeekToOrdinal(0));
rowid_t start_row_idx = 0;
while (iter->HasNext()) {
int batch_size = prng->Uniform(kMaxBatchSize) + 1;
- SCOPED_TRACE(strings::Substitute("APPLY: batch starting at $0 ($1 rows)",
+ SCOPED_TRACE(strings::Substitute("batch starting at $0 ($1 rows)",
start_row_idx, batch_size));
- ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_APPLY));
+ ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_APPLY |
+ DeltaIterator::PREPARE_FOR_COLLECT));
// Test ApplyDeletes: the selection vector is all true and a row is unset
// if the last relevant update deleted it.
@@ -851,18 +851,6 @@ void RunDeltaFuzzTest(const DeltaStore& store,
<< "\nActual column block: " << actual_scb.ToString();
}
- start_row_idx += batch_size;
- }
-
- // Now run PREPARE_FOR_COLLECT method tests.
- ASSERT_OK(iter->SeekToOrdinal(0));
- start_row_idx = 0;
- while (iter->HasNext()) {
- int batch_size = prng->Uniform(kMaxBatchSize - 1) + 1;
- SCOPED_TRACE(strings::Substitute("COLLECT: batch starting at $0 ($1 rows)",
- start_row_idx, batch_size));
- ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_COLLECT));
-
// Test CollectMutations: all relevant updates are returned.
{
Arena arena(1024);
@@ -917,6 +905,7 @@ void RunDeltaFuzzTest(const DeltaStore& store,
ASSERT_EQ(expected_deltas[j].second, actual_deltas[j].cell);
}
}
+
start_row_idx += batch_size;
}
}
[4/5] kudu git commit: (04/05) delta_store: support iteration with
is_deleted virtual column
Posted by mp...@apache.org.
(04/05) delta_store: support iteration with is_deleted virtual column
This patch adds support to delta stores (i.e. DMS and delta files) for
iterating with an IS_DELETED virtual column. A diff scan will include
IS_DELETED in the projection so that clients can tell if a row was deleted
in the requested time range.
Change-Id: Ibafcae9f9f82a6c2d9a8afedb8fbf07d6e3069ec
Reviewed-on: http://gerrit.cloudera.org:8080/11859
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/c931e570
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/c931e570
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/c931e570
Branch: refs/heads/master
Commit: c931e57076d2cd9a0bcc6adf6b59a21a37116b3e
Parents: 9f4657a
Author: Adar Dembo <ad...@cloudera.com>
Authored: Thu Nov 1 21:04:38 2018 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Fri Dec 7 03:45:28 2018 +0000
----------------------------------------------------------------------
src/kudu/common/schema.h | 20 +++++++++++++
src/kudu/tablet/delta_store.cc | 32 +++++++++++++++++++++
src/kudu/tablet/delta_store.h | 9 ++++++
src/kudu/tablet/deltafile-test.cc | 3 +-
src/kudu/tablet/memrowset.cc | 18 +-----------
src/kudu/tablet/memrowset.h | 5 +---
src/kudu/tablet/tablet-test-util.h | 51 ++++++++++++++++++++++++++++++---
7 files changed, 112 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/src/kudu/common/schema.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/schema.h b/src/kudu/common/schema.h
index 64283f0..4c19969 100644
--- a/src/kudu/common/schema.h
+++ b/src/kudu/common/schema.h
@@ -862,6 +862,26 @@ class Schema {
// Should be used when allocated on the heap.
size_t memory_footprint_including_this() const;
+ // Returns the column index for the first IS_DELETED virtual column in the
+ // schema, or kColumnNotFound if one cannot be found.
+ //
+ // The virtual column must not be nullable and must have a read default value.
+ // The process will crash if these constraints are not met.
+ int find_first_is_deleted_virtual_column() const {
+ for (int idx = 0; idx < num_columns(); idx++) {
+ const auto& col = column(idx);
+ if (col.type_info()->type() == IS_DELETED) {
+ // Enforce some properties on the virtual column that simplify our
+ // implementation.
+ DCHECK(!col.is_nullable());
+ DCHECK(col.has_read_default());
+
+ return idx;
+ }
+ }
+ return kColumnNotFound;
+ }
+
private:
// Return a stringified version of the first 'num_columns' columns of the
// row.
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/src/kudu/tablet/delta_store.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.cc b/src/kudu/tablet/delta_store.cc
index 4f97476..636ad8a 100644
--- a/src/kudu/tablet/delta_store.cc
+++ b/src/kudu/tablet/delta_store.cc
@@ -32,6 +32,7 @@
#include "kudu/common/schema.h"
#include "kudu/common/timestamp.h"
#include "kudu/common/types.h"
+#include "kudu/gutil/port.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/strcat.h"
#include "kudu/gutil/strings/substitute.h"
@@ -65,6 +66,7 @@ string DeltaKeyAndUpdate::Stringify(DeltaType type, const Schema& schema, bool p
template<class Traits>
DeltaPreparer<Traits>::DeltaPreparer(RowIteratorOptions opts)
: opts_(std::move(opts)),
+ projection_vc_is_deleted_idx_(opts_.projection->find_first_is_deleted_virtual_column()),
cur_prepared_idx_(0),
prev_prepared_idx_(0),
prepared_flags_(DeltaIterator::PREPARE_NONE),
@@ -232,6 +234,30 @@ Status DeltaPreparer<Traits>::ApplyUpdates(size_t col_to_apply, ColumnBlock* dst
DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, dst->nrows());
+ // Special handling for the IS_DELETED virtual column: convert 'deleted_' and
+ // 'reinserted_' into true and false cell values.
+ if (col_to_apply == projection_vc_is_deleted_idx_) {
+ // See ApplyDeletes() to understand why we adjust the virtual column's value
+ // for both deleted and reinserted rows.
+ for (const auto& row_id : deleted_) {
+ uint32_t idx_in_block = row_id - prev_prepared_idx_;
+ if (filter.IsRowSelected(idx_in_block)) {
+ ColumnBlock::Cell cell = dst->cell(idx_in_block);
+ UnalignedStore(cell.mutable_ptr(), true);
+ }
+ }
+
+ for (const auto& row_id : reinserted_) {
+ uint32_t idx_in_block = row_id - prev_prepared_idx_;
+ if (filter.IsRowSelected(idx_in_block)) {
+ ColumnBlock::Cell cell = dst->cell(idx_in_block);
+ UnalignedStore(cell.mutable_ptr(), false);
+ }
+ }
+
+ return Status::OK();
+ }
+
const ColumnSchema* col_schema = &opts_.projection->column(col_to_apply);
for (const ColumnUpdate& cu : updates_by_col_[col_to_apply]) {
int32_t idx_in_block = cu.row_id - prev_prepared_idx_;
@@ -252,6 +278,12 @@ Status DeltaPreparer<Traits>::ApplyDeletes(SelectionVector* sel_vec) {
DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, sel_vec->nrows());
+ // To understand why we must adjust sel_vec for both deleted_ and reinserted_,
+ // consider that DeltaIterators are often used en masse (i.e. via
+ // DeltaIteratorMerger). In such cases, it's possible for one DeltaPreparer to
+ // delete a row and for the next to reinsert it. Given that ApplyDeletes is
+ // called on each DeltaPreparer in order, we must "twiddle" sel_vec in either
+ // direction in order for the row's bit to hold the correct state at the end.
for (const auto& row_id : deleted_) {
uint32_t idx_in_block = row_id - prev_prepared_idx_;
sel_vec->SetRowUnselected(idx_in_block);
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/src/kudu/tablet/delta_store.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.h b/src/kudu/tablet/delta_store.h
index cd359c4..1f5f82b 100644
--- a/src/kudu/tablet/delta_store.h
+++ b/src/kudu/tablet/delta_store.h
@@ -351,6 +351,10 @@ class DeltaPreparer : public PreparedDeltas {
// Options with which the DeltaPreparer's iterator was constructed.
const RowIteratorOptions opts_;
+ // The index of the first IS_DELETED virtual column in the projection schema,
+ // or kColumnNotFound if one doesn't exist.
+ const int projection_vc_is_deleted_idx_;
+
// The row index at which the most recent batch preparation ended.
rowid_t cur_prepared_idx_;
@@ -372,6 +376,11 @@ class DeltaPreparer : public PreparedDeltas {
};
typedef std::deque<ColumnUpdate> UpdatesForColumn;
std::vector<UpdatesForColumn> updates_by_col_;
+
+ // A row whose last relevant mutation was DELETE (or REINSERT).
+ //
+ // These deques are disjoint; a row that was both deleted and reinserted will
+ // not be in either.
std::deque<rowid_t> deleted_;
std::deque<rowid_t> reinserted_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/src/kudu/tablet/deltafile-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile-test.cc b/src/kudu/tablet/deltafile-test.cc
index 1226363..aed693c 100644
--- a/src/kudu/tablet/deltafile-test.cc
+++ b/src/kudu/tablet/deltafile-test.cc
@@ -495,7 +495,8 @@ TYPED_TEST(DeltaTypeTestDeltaFile, BenchmarkPrepareAndApply) {
for (int i = 0; i < kNumScans; i++) {
// Create a random projection with kNumColumnsToScan columns.
- Schema projection = GetRandomProjection(schema, &prng, kNumColumnsToScan);
+ Schema projection = GetRandomProjection(schema, &prng, kNumColumnsToScan,
+ AllowIsDeleted::NO);
// Create an iterator at a randomly chosen timestamp.
Timestamp ts = Timestamp(prng.Uniform(kMaxTimestamp));
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/src/kudu/tablet/memrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc
index da107af..2045d18 100644
--- a/src/kudu/tablet/memrowset.cc
+++ b/src/kudu/tablet/memrowset.cc
@@ -29,13 +29,11 @@
#include "kudu/codegen/compilation_manager.h"
#include "kudu/codegen/row_projector.h"
#include "kudu/common/columnblock.h"
-#include "kudu/common/common.pb.h"
#include "kudu/common/encoded_key.h"
#include "kudu/common/row.h"
#include "kudu/common/row_changelist.h"
#include "kudu/common/rowblock.h"
#include "kudu/common/scan_spec.h"
-#include "kudu/common/types.h"
#include "kudu/consensus/log_anchor_registry.h"
#include "kudu/consensus/opid.pb.h"
#include "kudu/gutil/dynamic_annotations.h"
@@ -396,27 +394,13 @@ MemRowSet::Iterator::Iterator(const std::shared_ptr<const MemRowSet>& mrs,
projector_(
GenerateAppropriateProjector(&mrs->schema_nonvirtual(), opts_.projection)),
delta_projector_(&mrs->schema_nonvirtual(), opts_.projection),
+ projection_vc_is_deleted_idx_(opts_.projection->find_first_is_deleted_virtual_column()),
state_(kUninitialized) {
// TODO(todd): various code assumes that a newly constructed iterator
// is pointed at the beginning of the dataset. This causes a redundant
// seek. Could make this lazy instead, or change the semantics so that
// a seek is required (probably the latter)
iter_->SeekToStart();
-
- // Find the first IS_DELETED virtual column, if one exists.
- projection_vc_is_deleted_idx_ = Schema::kColumnNotFound;
- for (int i = 0; i < opts_.projection->num_columns(); i++) {
- const auto& col = opts_.projection->column(i);
- if (col.type_info()->type() == IS_DELETED) {
- // Enforce some properties on the virtual column that simplify our
- // implementation.
- DCHECK(!col.is_nullable());
- DCHECK(col.has_read_default());
-
- projection_vc_is_deleted_idx_ = i;
- break;
- }
- }
}
MemRowSet::Iterator::~Iterator() {}
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/src/kudu/tablet/memrowset.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.h b/src/kudu/tablet/memrowset.h
index dc382d8..5803a7e 100644
--- a/src/kudu/tablet/memrowset.h
+++ b/src/kudu/tablet/memrowset.h
@@ -595,10 +595,7 @@ class MemRowSet::Iterator : public RowwiseIterator {
// The index of the first IS_DELETED virtual column in the projection schema,
// or kColumnNotFound if one doesn't exist.
- //
- // The virtual column must not be nullable and must have a read default value.
- // The process will crash if these constraints are not met.
- int projection_vc_is_deleted_idx_;
+ const int projection_vc_is_deleted_idx_;
// Temporary buffer used for RowChangeList projection.
faststring delta_buf_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/c931e570/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 f43e40d..9224cb0 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -609,6 +609,13 @@ static inline std::vector<size_t> GetRandomIntegerSequence(
return std::vector<size_t>(integers.begin(), integers.end());
}
+// Generates random deltas conforming to 'schema' and stores them in 'mirror'.
+//
+// 'row_range' and 'ts_range' constrain the DeltaKeys used in the created deltas.
+//
+// If 'allow_reinserts' is true, REINSERT deltas may also be generated.
+// Otherwise, a row won't receive any more deltas after a DELETE has been
+// generated for it.
template <typename T>
void CreateRandomDeltas(const Schema& schema,
Random* prng,
@@ -703,9 +710,22 @@ void CreateRandomDeltas(const Schema& schema,
}
}
+// Create a random projection conforming to 'schema'.
+//
+// 'max_cols_to_project' defines the maximum number of columns that should be
+// allowed into the projection; the actual number of columns is randomly
+// generated.
+//
+// If 'allow' is true, an IS_DELETED virtual column may be randomly added to
+// the projection.
+enum class AllowIsDeleted {
+ YES,
+ NO
+};
static inline Schema GetRandomProjection(const Schema& schema,
Random* prng,
- size_t max_cols_to_project) {
+ size_t max_cols_to_project,
+ AllowIsDeleted allow) {
// Set up the projection.
auto idxs_to_project = GetRandomIntegerSequence(prng,
schema.num_columns(),
@@ -716,6 +736,14 @@ static inline Schema GetRandomProjection(const Schema& schema,
projected_cols.emplace_back(schema.column(idx));
projected_col_ids.emplace_back(schema.column_id(idx));
}
+
+ // Add a IS_DELETED virtual column some of the time.
+ if (allow == AllowIsDeleted::YES && prng->Uniform(10) == 0) {
+ bool read_default = false;
+ projected_cols.emplace_back("is_deleted", IS_DELETED, /*is_nullable=*/ false,
+ &read_default);
+ projected_col_ids.emplace_back(schema.max_col_id() + 1);
+ }
return Schema(projected_cols, projected_col_ids, 0);
}
@@ -859,7 +887,11 @@ void RunDeltaFuzzTest(const DeltaStore& store,
// Create and initialize the iterator. If none iterator is returned, it's
// because all deltas in 'store' were irrelevant; verify this.
- Schema projection = GetRandomProjection(*mirror->schema(), prng, kMaxColsToProject);
+ Schema projection = GetRandomProjection(*mirror->schema(), prng, kMaxColsToProject,
+ lower_ts ? AllowIsDeleted::YES :
+ AllowIsDeleted::NO);
+ size_t projection_vc_is_deleted_idx =
+ projection.find_first_is_deleted_virtual_column();
SCOPED_TRACE(strings::Substitute("Projection $0", projection.ToString()));
RowIteratorOptions opts;
opts.projection = &projection;
@@ -935,8 +967,19 @@ void RunDeltaFuzzTest(const DeltaStore& store,
actual_scb[k] = 0;
}
const SelectionVector& filter = lower_ts ? actual_selected : actual_deleted;
- ASSERT_OK(mirror->ApplyUpdates(*opts.projection, lower_ts, upper_ts,
- start_row_idx, j, &expected_scb, filter));
+ if (j == projection_vc_is_deleted_idx) {
+ // Reconstruct the expected IS_DELETED state using 'actual_selected'
+ // and 'actual_deleted', which we've already verified above.
+ DCHECK(lower_ts);
+ for (int k = 0; k < batch_size; k++) {
+ if (actual_selected.IsRowSelected(k)) {
+ expected_scb[k] = !actual_deleted.IsRowSelected(k);
+ }
+ }
+ } else {
+ ASSERT_OK(mirror->ApplyUpdates(*opts.projection, lower_ts, upper_ts,
+ start_row_idx, j, &expected_scb, filter));
+ }
ASSERT_OK(iter->ApplyUpdates(j, &actual_scb, filter));
ASSERT_EQ(expected_scb, actual_scb)
<< "Expected column block: " << expected_scb.ToString()
[3/5] kudu git commit: (03/05) delta_store: support iteration with
snap_to_exclude
Posted by mp...@apache.org.
(03/05) delta_store: support iteration with snap_to_exclude
This patch changes the delta stores (DMS and delta files) to respect
snap_to_exclude during iteration. The key changes are:
- The introduction of the "selection" criteria, a new delta relevancy
formula for determining whether a delta applies to a scan with both
snap_to_exclude and snap_to_include. The existing "application" criteria
was formalized and moved into delta_relevancy.h. There was also a
non-trivial change to DeltaFileReader::IsRelevantForSnapshot() to use both
criterias when culling entire delta files.
- A new SelectUpdates() method for using the selection criteria on a batch
of prepared deltas. SelectUpdates() requires new in-memory state, the
creation of which is gated behind a new PREPARE_FOR_SELECT flag so as not
to affect regular scans.
- Updates to the delta fuzz testing logic to test iterators with two
timestamps, and to provide SelectUpdates() coverage.
A future patch will modify DeltaApplier to use SelectUpdates for diff scans.
Change-Id: I7811e185fef270f40fdbbb38f491eee8b4aa043c
Reviewed-on: http://gerrit.cloudera.org:8080/11858
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/9f4657ab
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/9f4657ab
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/9f4657ab
Branch: refs/heads/master
Commit: 9f4657ab56366df541eddce12850726038d95fd0
Parents: d927675
Author: Adar Dembo <ad...@cloudera.com>
Authored: Thu Nov 1 17:07:18 2018 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Fri Dec 7 03:45:13 2018 +0000
----------------------------------------------------------------------
src/kudu/tablet/delta_iterator_merger.cc | 7 +
src/kudu/tablet/delta_iterator_merger.h | 2 +
src/kudu/tablet/delta_relevancy.h | 181 ++++++++++++++++++++++++++
src/kudu/tablet/delta_store.cc | 132 ++++++++++++-------
src/kudu/tablet/delta_store.h | 28 +++-
src/kudu/tablet/deltafile.cc | 41 ++++--
src/kudu/tablet/deltafile.h | 8 +-
src/kudu/tablet/deltamemstore.cc | 5 +-
src/kudu/tablet/deltamemstore.h | 2 +
src/kudu/tablet/tablet-test-util.cc | 17 ++-
src/kudu/tablet/tablet-test-util.h | 142 ++++++++++++++++----
11 files changed, 473 insertions(+), 92 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/delta_iterator_merger.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_iterator_merger.cc b/src/kudu/tablet/delta_iterator_merger.cc
index f6f5840..58d14da 100644
--- a/src/kudu/tablet/delta_iterator_merger.cc
+++ b/src/kudu/tablet/delta_iterator_merger.cc
@@ -79,6 +79,13 @@ Status DeltaIteratorMerger::ApplyDeletes(SelectionVector* sel_vec) {
return Status::OK();
}
+Status DeltaIteratorMerger::SelectUpdates(SelectionVector* sel_vec) {
+ for (const unique_ptr<DeltaIterator>& iter : iters_) {
+ RETURN_NOT_OK(iter->SelectUpdates(sel_vec));
+ }
+ return Status::OK();
+}
+
Status DeltaIteratorMerger::CollectMutations(vector<Mutation*>* dst, Arena* arena) {
for (const unique_ptr<DeltaIterator> &iter : iters_) {
RETURN_NOT_OK(iter->CollectMutations(dst, arena));
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/delta_iterator_merger.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_iterator_merger.h b/src/kudu/tablet/delta_iterator_merger.h
index a8efeb6..c5a36e8 100644
--- a/src/kudu/tablet/delta_iterator_merger.h
+++ b/src/kudu/tablet/delta_iterator_merger.h
@@ -67,6 +67,8 @@ class DeltaIteratorMerger : public DeltaIterator {
Status ApplyDeletes(SelectionVector* sel_vec) override;
+ Status SelectUpdates(SelectionVector* sel_vec) override;
+
Status CollectMutations(std::vector<Mutation*>* dst, Arena* arena) override;
Status FilterColumnIdsAndCollectDeltas(const std::vector<ColumnId>& col_ids,
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/delta_relevancy.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_relevancy.h b/src/kudu/tablet/delta_relevancy.h
new file mode 100644
index 0000000..69640ac
--- /dev/null
+++ b/src/kudu/tablet/delta_relevancy.h
@@ -0,0 +1,181 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "kudu/common/timestamp.h"
+#include "kudu/tablet/delta_key.h"
+#include "kudu/tablet/mvcc.h"
+
+namespace kudu {
+namespace tablet {
+
+// Functions that evaluate the relevancy of deltas against a snapshot or pair
+// of snapshots.
+//
+// When constructing an iterator tree, one must specify either one or two MVCC
+// snapshots. The first always serves as an upper bound, ensuring that the data
+// may be further mutated by writers without affecting the results of the
+// iteration. If present, the second serves as a lower bound, excluding any
+// deltas that were committed before it in time.
+//
+// Together, the snapshots affect the behavior of deltas encountered during the
+// iteration. Various criteria is used to determine whether a particular delta
+// is relevant to the iteration; which criteria is used depends on the context.
+//
+// Selection criteria
+// ==================
+// The selection (or "select") criteria determines whether the particular delta
+// should cause its row to be included in the iteration. It's only applicable to
+// two snapshots. In detail, regardless of delta type, if the delta's timestamp
+// is "between" the two snapshots (i.e. not committed in the lower bound
+// snapshot but committed in the upper bound snapshot), its row should be included.
+//
+// Application criteria
+// ====================
+// The application (or "apply") criteria determines whether the logical contents
+// of the UPDATE, DELETE, or REINSERT in a particular delta should be applied to
+// a row block during a scan. It only considers the first snapshot. In detail:
+// - For REDO deltas: if the delta's timestamp is committed in the snapshot,
+// the mutation should be applied.
+// - For UNDO deltas: if the delta's timestamp is not committed in the snapshot,
+// the mutation should be applied.
+
+// Returns whether a delta at 'delta_ts' is relevant under the apply criteria to 'snap'.
+template<DeltaType Type>
+inline bool IsDeltaRelevantForApply(const MvccSnapshot& snap,
+ const Timestamp& delta_ts) {
+ bool ignored;
+ return IsDeltaRelevantForApply<Type>(snap, delta_ts, &ignored);
+}
+
+// A variant of IsDeltaRelevantForApply that, if the delta is not relevant,
+// further checks whether any remaining deltas for this row can be skipped; this
+// is an optimization and not necessary for correctness.
+template<DeltaType Type>
+inline bool IsDeltaRelevantForApply(const MvccSnapshot& snap,
+ const Timestamp& delta_ts,
+ bool* finished_row);
+
+template<>
+inline bool IsDeltaRelevantForApply<REDO>(const MvccSnapshot& snap,
+ const Timestamp& delta_ts,
+ bool* finished_row) {
+ *finished_row = false;
+ if (snap.IsCommitted(delta_ts)) {
+ return true;
+ }
+ if (!snap.MayHaveCommittedTransactionsAtOrAfter(delta_ts)) {
+ // REDO deltas are sorted first in ascending row ordinal order, then in
+ // ascending timestamp order. Thus, if we know that there are no more
+ // committed transactions whose timestamps are >= 'delta_ts', we know that
+ // any future deltas belonging to this row aren't relevant (as per the apply
+ // criteria, REDOs are relevant if they are committed in the snapshot), and
+ // we can skip to the next row.
+ *finished_row = true;
+ }
+ return false;
+}
+
+template<>
+inline bool IsDeltaRelevantForApply<UNDO>(const MvccSnapshot& snap,
+ const Timestamp& delta_ts,
+ bool* finished_row) {
+ *finished_row = false;
+ if (!snap.IsCommitted(delta_ts)) {
+ return true;
+ }
+ if (!snap.MayHaveUncommittedTransactionsAtOrBefore(delta_ts)) {
+ // UNDO deltas are sorted first in ascending row ordinal order, then in
+ // descending timestamp order. Thus, if we know that there are no more
+ // uncommitted transactions whose timestamps are <= 'delta_ts', we know that
+ // any future deltas belonging to this row aren't relevant (as per the apply
+ // criteria, UNDOs are relevant if they are uncommitted in the snapshot),
+ // and we can skip to the next row.
+ *finished_row = true;
+ }
+ return false;
+}
+
+// Returns whether deltas within the time range 'delta_ts_start' to
+// 'delta_ts_end' are relevant under the select criteria to 'snap_start' and 'snap_end'.
+inline bool IsDeltaRelevantForSelect(const MvccSnapshot& snap_start,
+ const MvccSnapshot& snap_end,
+ const Timestamp& delta_ts_start,
+ const Timestamp& delta_ts_end) {
+ return !snap_start.IsCommitted(delta_ts_end) &&
+ snap_end.IsCommitted(delta_ts_start);
+}
+
+// A variant of IsDeltaRelevantForSelect that operates on a single delta's
+// timestamp given by 'delta_ts', and if the delta is not relevant, further
+// checks whether any remaining deltas for this row can be skipped; this is an
+// optimization and not necessary for correctness.
+template<DeltaType Type>
+inline bool IsDeltaRelevantForSelect(const MvccSnapshot& snap_start,
+ const MvccSnapshot& snap_end,
+ const Timestamp& delta_ts,
+ bool* finished_row);
+
+template<>
+inline bool IsDeltaRelevantForSelect<REDO>(const MvccSnapshot& snap_start,
+ const MvccSnapshot& snap_end,
+ const Timestamp& delta_ts,
+ bool* finished_row) {
+ *finished_row = false;
+ if (snap_start.IsCommitted(delta_ts)) {
+ // No short-circuit available here; because REDO deltas for a given row are
+ // sorted in ascending timestamp order, the next REDO may be uncommitted in
+ // 'snap_start'.
+ return false;
+ }
+ if (!snap_end.IsCommitted(delta_ts)) {
+ if (!snap_end.MayHaveCommittedTransactionsAtOrAfter(delta_ts)) {
+ // But if 'delta_ts' is not committed in 'snap_end', all future REDOs may
+ // also be uncommitted in 'snap_end'.
+ *finished_row = true;
+ }
+ return false;
+ }
+ return true;
+}
+
+template<>
+inline bool IsDeltaRelevantForSelect<UNDO>(const MvccSnapshot& snap_start,
+ const MvccSnapshot& snap_end,
+ const Timestamp& delta_ts,
+ bool* finished_row) {
+ *finished_row = false;
+ if (!snap_end.IsCommitted(delta_ts)) {
+ // No short-circuit available here; because UNDO deltas for a given row are
+ // sorted in descending timestamp order, the next UNDO may be committed in
+ // 'snap_end'.
+ return false;
+ }
+ if (snap_start.IsCommitted(delta_ts)) {
+ if (!snap_start.MayHaveUncommittedTransactionsAtOrBefore(delta_ts)) {
+ // But if 'delta_ts' is committed in 'snap_start', all future UNDOs may
+ // also be committed in 'snap_start'.
+ *finished_row = true;
+ }
+ return false;
+ }
+ return true;
+}
+
+} // namespace tablet
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/delta_store.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.cc b/src/kudu/tablet/delta_store.cc
index bb7a91f..4f97476 100644
--- a/src/kudu/tablet/delta_store.cc
+++ b/src/kudu/tablet/delta_store.cc
@@ -35,6 +35,7 @@
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/strcat.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/tablet/delta_relevancy.h"
#include "kudu/tablet/delta_stats.h"
#include "kudu/tablet/deltafile.h"
#include "kudu/tablet/mutation.h"
@@ -51,47 +52,6 @@ using std::string;
using std::vector;
using strings::Substitute;
-namespace {
-
-// Returns whether a mutation at 'ts' is relevant under 'snap'.
-//
-// If not relevant, further checks whether any remaining deltas for this row can
-// be skipped; this is an optimization and not necessary for correctness.
-template<DeltaType Type>
-bool IsDeltaRelevant(const MvccSnapshot& snap,
- const Timestamp& ts,
- bool* finished_row);
-
-template<>
-bool IsDeltaRelevant<REDO>(const MvccSnapshot& snap,
- const Timestamp& ts,
- bool* finished_row) {
- *finished_row = false;
- if (!snap.IsCommitted(ts)) {
- if (!snap.MayHaveCommittedTransactionsAtOrAfter(ts)) {
- *finished_row = true;
- }
- return false;
- }
- return true;
-}
-
-template<>
-bool IsDeltaRelevant<UNDO>(const MvccSnapshot& snap,
- const Timestamp& ts,
- bool* finished_row) {
- *finished_row = false;
- if (snap.IsCommitted(ts)) {
- if (!snap.MayHaveUncommittedTransactionsAtOrBefore(ts)) {
- *finished_row = true;
- }
- return false;
- }
- return true;
-}
-
-} // anonymous namespace
-
string DeltaKeyAndUpdate::Stringify(DeltaType type, const Schema& schema, bool pad_key) const {
return StrCat(Substitute("($0 delta key=$2, change_list=$1)",
DeltaType_Name(type),
@@ -119,8 +79,18 @@ void DeltaPreparer<Traits>::Seek(rowid_t row_idx) {
}
template<class Traits>
-void DeltaPreparer<Traits>::Start(int prepare_flags) {
+void DeltaPreparer<Traits>::Start(size_t nrows, int prepare_flags) {
DCHECK_NE(prepare_flags, DeltaIterator::PREPARE_NONE);
+
+ if (prepare_flags & DeltaIterator::PREPARE_FOR_SELECT) {
+ DCHECK(opts_.snap_to_exclude);
+
+ // Ensure we have a selection vector at least 'nrows' long.
+ if (!selected_ || selected_->nrows() < nrows) {
+ selected_.reset(new SelectionVector(nrows));
+ }
+ selected_->SetAllFalse();
+ }
prepared_flags_ = prepare_flags;
if (updates_by_col_.empty()) {
updates_by_col_.resize(opts_.projection->num_columns());
@@ -132,6 +102,13 @@ void DeltaPreparer<Traits>::Start(int prepare_flags) {
reinserted_.clear();
prepared_deltas_.clear();
deletion_state_ = UNKNOWN;
+
+ if (VLOG_IS_ON(3)) {
+ string snap_to_exclude = opts_.snap_to_exclude ?
+ opts_.snap_to_exclude->ToString() : "INF";
+ VLOG(3) << "Starting batch for [" << snap_to_exclude << ","
+ << opts_.snap_to_include.ToString() << ")";
+ }
}
template<class Traits>
@@ -139,17 +116,53 @@ void DeltaPreparer<Traits>::Finish(size_t nrows) {
MaybeProcessPreviousRowChange(boost::none);
prev_prepared_idx_ = cur_prepared_idx_;
cur_prepared_idx_ += nrows;
+
+ if (VLOG_IS_ON(3)) {
+ string snap_to_exclude = opts_.snap_to_exclude ?
+ opts_.snap_to_exclude->ToString() : "INF";
+ VLOG(3) << "Finishing batch for [" << snap_to_exclude << ","
+ << opts_.snap_to_include.ToString() << ")";
+ }
}
template<class Traits>
Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* finished_row) {
- if (!IsDeltaRelevant<Traits::kType>(opts_.snap_to_include,
- key.timestamp(), finished_row)) {
- return Status::OK();
- }
MaybeProcessPreviousRowChange(key.row_idx());
- if (prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY) {
+ VLOG(4) << "Considering delta " << key.ToString() << ": "
+ << RowChangeList(val).ToString(*opts_.projection);
+
+ // Different preparations may use different criteria for delta relevancy. Each
+ // criteria offers a short-circuit when processing of the current row is known
+ // to be finished, but that short-circuit can only be used if we're not also
+ // handling a preparation with a different criteria.
+
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_SELECT) {
+ bool finished_row_for_select;
+ if (IsDeltaRelevantForSelect<Traits::kType>(*opts_.snap_to_exclude,
+ opts_.snap_to_include,
+ key.timestamp(),
+ &finished_row_for_select)) {
+ selected_->SetRowSelected(key.row_idx() - cur_prepared_idx_);
+ }
+
+ if (finished_row_for_select &&
+ !(prepared_flags_ & ~DeltaIterator::PREPARE_FOR_SELECT)) {
+ *finished_row = true;
+ }
+ }
+
+ // Apply and collect use the same relevancy criteria.
+ bool relevant_for_apply_or_collect = false;
+ bool finished_row_for_apply_or_collect = false;
+ if (prepared_flags_ & (DeltaIterator::PREPARE_FOR_APPLY |
+ DeltaIterator::PREPARE_FOR_COLLECT)) {
+ relevant_for_apply_or_collect = IsDeltaRelevantForApply<Traits::kType>(
+ opts_.snap_to_include, key.timestamp(), &finished_row_for_apply_or_collect);
+ }
+
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY &&
+ relevant_for_apply_or_collect) {
RowChangeListDecoder decoder((RowChangeList(val)));
if (Traits::kInitializeDecodersWithSafetyChecks) {
RETURN_NOT_OK(decoder.Init());
@@ -194,13 +207,21 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
}
}
}
- if (prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT) {
+
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT &&
+ relevant_for_apply_or_collect) {
PreparedDelta d;
d.key = key;
d.val = val;
prepared_deltas_.emplace_back(d);
}
+ if (finished_row_for_apply_or_collect &&
+ !(prepared_flags_ & ~(DeltaIterator::PREPARE_FOR_APPLY |
+ DeltaIterator::PREPARE_FOR_COLLECT))) {
+ *finished_row = true;
+ }
+
last_added_idx_ = key.row_idx();
return Status::OK();
}
@@ -245,6 +266,21 @@ Status DeltaPreparer<Traits>::ApplyDeletes(SelectionVector* sel_vec) {
}
template<class Traits>
+Status DeltaPreparer<Traits>::SelectUpdates(SelectionVector* sel_vec) {
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_SELECT);
+ DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, sel_vec->nrows());
+
+ // SelectUpdates() is additive: it should never exclude rows, only include them.
+ for (rowid_t idx = 0; idx < sel_vec->nrows(); idx++) {
+ if (selected_->IsRowSelected(idx)) {
+ sel_vec->SetRowSelected(idx);
+ }
+ }
+
+ return Status::OK();
+}
+
+template<class Traits>
Status DeltaPreparer<Traits>::CollectMutations(vector<Mutation*>* dst, Arena* arena) {
DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, dst->size());
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/delta_store.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.h b/src/kudu/tablet/delta_store.h
index 955f4a7..cd359c4 100644
--- a/src/kudu/tablet/delta_store.h
+++ b/src/kudu/tablet/delta_store.h
@@ -155,6 +155,14 @@ class PreparedDeltas {
// Deltas must have been prepared with the flag PREPARE_FOR_APPLY.
virtual Status ApplyDeletes(SelectionVector* sel_vec) = 0;
+ // Updates the given selection vector to reflect the snapshotted updates.
+ //
+ // Rows which have been updated or deleted in the associated MVCC snapshot are
+ // set to 1 in the selection vector so that they show up in the output.
+ //
+ // Deltas must have been prepared with the flag PREPARE_FOR_SELECT.
+ virtual Status SelectUpdates(SelectionVector* sel_vec) = 0;
+
// Collects the mutations associated with each row in the current prepared batch.
//
// Each entry in the vector will be treated as a singly linked list of Mutation
@@ -226,7 +234,14 @@ class DeltaIterator : public PreparedDeltas {
// in the order that they were loaded from the backing store.
//
// On success, CollectMutations and FilterColumnIdsAndCollectDeltas will be callable.
- PREPARE_FOR_COLLECT = 1 << 1
+ PREPARE_FOR_COLLECT = 1 << 1,
+
+ // Prepare a batch of deltas for selecting. All deltas in the batch will be
+ // decoded, and a data structure describing which rows had deltas will be
+ // populated.
+ //
+ // On success, SelectUpdates will be callable.
+ PREPARE_FOR_SELECT = 1 << 2
};
virtual Status PrepareBatch(size_t nrows, int prepare_flags) = 0;
@@ -281,7 +296,7 @@ class DeltaPreparer : public PreparedDeltas {
// on the part of a DeltaIterator.
//
// Call at the beginning of DeltaIterator::PrepareBatch.
- void Start(int prepare_flags);
+ void Start(size_t nrows, int prepare_flags);
// Updates internal state to reflect the end of delta batch preparation on the
// part of a DeltaIterator.
@@ -305,6 +320,8 @@ class DeltaPreparer : public PreparedDeltas {
Status ApplyDeletes(SelectionVector* sel_vec) override;
+ Status SelectUpdates(SelectionVector* sel_vec) override;
+
Status CollectMutations(std::vector<Mutation*>* dst, Arena* arena) override;
Status FilterColumnIdsAndCollectDeltas(const std::vector<ColumnId>& col_ids,
@@ -359,6 +376,9 @@ class DeltaPreparer : public PreparedDeltas {
std::deque<rowid_t> reinserted_;
// The deletion state of the row last processed by AddDelta().
+ //
+ // As a row's DELETEs and REINSERTs are processed, the deletion state
+ // alternates between the values below.
enum RowDeletionState {
UNKNOWN,
DELETED,
@@ -374,6 +394,10 @@ class DeltaPreparer : public PreparedDeltas {
};
std::deque<PreparedDelta> prepared_deltas_;
+ // State when prepared_for_ & PREPARED_FOR_SELECT
+ // ------------------------------------------------------------
+ std::unique_ptr<SelectionVector> selected_;
+
DISALLOW_COPY_AND_ASSIGN(DeltaPreparer);
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/deltafile.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.cc b/src/kudu/tablet/deltafile.cc
index 7ca6719..30bb01b 100644
--- a/src/kudu/tablet/deltafile.cc
+++ b/src/kudu/tablet/deltafile.cc
@@ -44,6 +44,7 @@
#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/tablet/delta_relevancy.h"
#include "kudu/tablet/mvcc.h"
#include "kudu/tablet/rowset.h"
#include "kudu/tablet/tablet.pb.h"
@@ -285,20 +286,30 @@ Status DeltaFileReader::ReadDeltaStats() {
return Status::OK();
}
-bool DeltaFileReader::IsRelevantForSnapshot(const MvccSnapshot& snap) const {
+bool DeltaFileReader::IsRelevantForSnapshots(
+ const boost::optional<MvccSnapshot>& snap_to_exclude,
+ const MvccSnapshot& snap_to_include) const {
if (!init_once_.init_succeeded()) {
// If we're not initted, it means we have no delta stats and must
// assume that this file is relevant for every snapshot.
return true;
}
- if (delta_type_ == REDO) {
- return snap.MayHaveCommittedTransactionsAtOrAfter(delta_stats_->min_timestamp());
- }
- if (delta_type_ == UNDO) {
- return snap.MayHaveUncommittedTransactionsAtOrBefore(delta_stats_->max_timestamp());
+
+ // We don't know whether the caller's intent is to apply deltas, to select
+ // them, or both. As such, we must be conservative and assume 'both', which
+ // means the file is relevant if any relevancy criteria is true.
+ bool relevant = delta_type_ == REDO ?
+ IsDeltaRelevantForApply<REDO>(snap_to_include,
+ delta_stats_->min_timestamp()) :
+ IsDeltaRelevantForApply<UNDO>(snap_to_include,
+ delta_stats_->max_timestamp());
+ if (snap_to_exclude) {
+ // The select criteria is the same regardless of delta_type_.
+ relevant |= IsDeltaRelevantForSelect(*snap_to_exclude, snap_to_include,
+ delta_stats_->min_timestamp(),
+ delta_stats_->max_timestamp());
}
- LOG(DFATAL) << "Cannot reach here";
- return false;
+ return relevant;
}
Status DeltaFileReader::CloneForDebugging(FsManager* fs_manager,
@@ -313,7 +324,7 @@ Status DeltaFileReader::CloneForDebugging(FsManager* fs_manager,
Status DeltaFileReader::NewDeltaIterator(const RowIteratorOptions& opts,
DeltaIterator** iterator) const {
- if (IsRelevantForSnapshot(opts.snap_to_include)) {
+ if (IsRelevantForSnapshots(opts.snap_to_exclude, opts.snap_to_include)) {
if (VLOG_IS_ON(2)) {
if (!init_once_.init_succeeded()) {
TRACE_COUNTER_INCREMENT("delta_iterators_lazy_initted", 1);
@@ -429,11 +440,12 @@ Status DeltaFileIterator<Type>::SeekToOrdinal(rowid_t idx) {
// Finish the initialization of any lazily-initialized state.
RETURN_NOT_OK(dfr_->Init(preparer_.opts().io_context));
- // Check again whether this delta file is relevant given the snapshot
+ // Check again whether this delta file is relevant given the snapshots
// that we are querying. We did this already before creating the
// DeltaFileIterator, but due to lazy initialization, it's possible
// that we weren't able to check at that time.
- if (!dfr_->IsRelevantForSnapshot(preparer_.opts().snap_to_include)) {
+ if (!dfr_->IsRelevantForSnapshots(preparer_.opts().snap_to_exclude,
+ preparer_.opts().snap_to_include)) {
exhausted_ = true;
delta_blocks_.clear();
return Status::OK();
@@ -587,7 +599,7 @@ Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, int prepare_flags) {
#endif
prepared_ = true;
- preparer_.Start(prepare_flags);
+ preparer_.Start(nrows, prepare_flags);
RETURN_NOT_OK(AddDeltas(start_row, stop_row));
preparer_.Finish(nrows);
return Status::OK();
@@ -672,6 +684,11 @@ Status DeltaFileIterator<Type>::ApplyDeletes(SelectionVector* sel_vec) {
}
template<DeltaType Type>
+Status DeltaFileIterator<Type>::SelectUpdates(SelectionVector* sel_vec) {
+ return preparer_.SelectUpdates(sel_vec);
+}
+
+template<DeltaType Type>
Status DeltaFileIterator<Type>::CollectMutations(vector<Mutation*>* dst, Arena* arena) {
return preparer_.CollectMutations(dst, arena);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/deltafile.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.h b/src/kudu/tablet/deltafile.h
index 217dd4c..e330ce5 100644
--- a/src/kudu/tablet/deltafile.h
+++ b/src/kudu/tablet/deltafile.h
@@ -24,6 +24,7 @@
#include <string>
#include <vector>
+#include <boost/optional/optional.hpp>
#include <glog/logging.h>
#include "kudu/cfile/block_handle.h"
@@ -180,9 +181,10 @@ class DeltaFileReader : public DeltaStore,
}
// Returns true if this delta file may include any deltas which need to be
- // applied when scanning the given snapshot, or if the file has not yet
+ // applied when scanning the given snapshots, or if the file has not yet
// been fully initialized.
- bool IsRelevantForSnapshot(const MvccSnapshot& snap) const;
+ bool IsRelevantForSnapshots(const boost::optional<MvccSnapshot>& snap_to_exclude,
+ const MvccSnapshot& snap_to_include) const;
// Clone this DeltaFileReader for testing and validation purposes (such as
// while in DEBUG mode). The resulting object will not be Initted().
@@ -234,6 +236,8 @@ class DeltaFileIterator : public DeltaIterator {
Status ApplyDeletes(SelectionVector* sel_vec) override;
+ Status SelectUpdates(SelectionVector* sel_vec) override;
+
Status CollectMutations(std::vector<Mutation*>*dst, Arena* arena) override;
Status FilterColumnIdsAndCollectDeltas(const std::vector<ColumnId>& col_ids,
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/deltamemstore.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltamemstore.cc b/src/kudu/tablet/deltamemstore.cc
index a25c110..c2a8f0d 100644
--- a/src/kudu/tablet/deltamemstore.cc
+++ b/src/kudu/tablet/deltamemstore.cc
@@ -234,7 +234,7 @@ Status DMSIterator::PrepareBatch(size_t nrows, int prepare_flags) {
rowid_t start_row = preparer_.cur_prepared_idx();
rowid_t stop_row = start_row + nrows - 1;
- preparer_.Start(prepare_flags);
+ preparer_.Start(nrows, prepare_flags);
bool finished_row = false;
while (iter_->IsValid()) {
Slice key_slice, val;
@@ -284,6 +284,9 @@ Status DMSIterator::ApplyDeletes(SelectionVector* sel_vec) {
return preparer_.ApplyDeletes(sel_vec);
}
+Status DMSIterator::SelectUpdates(SelectionVector* sel_vec) {
+ return preparer_.SelectUpdates(sel_vec);
+}
Status DMSIterator::CollectMutations(vector<Mutation*>*dst, Arena* arena) {
return preparer_.CollectMutations(dst, arena);
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/deltamemstore.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltamemstore.h b/src/kudu/tablet/deltamemstore.h
index a5a92f7..3cc02a0 100644
--- a/src/kudu/tablet/deltamemstore.h
+++ b/src/kudu/tablet/deltamemstore.h
@@ -206,6 +206,8 @@ class DMSIterator : public DeltaIterator {
Status ApplyDeletes(SelectionVector* sel_vec) override;
+ Status SelectUpdates(SelectionVector* sel_vec) override;
+
Status CollectMutations(std::vector<Mutation*>* dst, Arena* arena) override;
Status FilterColumnIdsAndCollectDeltas(const std::vector<ColumnId>& col_ids,
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/src/kudu/tablet/tablet-test-util.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test-util.cc b/src/kudu/tablet/tablet-test-util.cc
index 0575e3b..5453c30 100644
--- a/src/kudu/tablet/tablet-test-util.cc
+++ b/src/kudu/tablet/tablet-test-util.cc
@@ -21,16 +21,29 @@ namespace kudu {
namespace tablet {
template<>
-bool MirroredDeltas<DeltaTypeSelector<REDO>>::IsDeltaRelevant(
+bool MirroredDeltas<DeltaTypeSelector<REDO>>::IsDeltaRelevantForApply(
Timestamp to_include, Timestamp ts) const {
return ts < to_include;
}
template<>
-bool MirroredDeltas<DeltaTypeSelector<UNDO>>::IsDeltaRelevant(
+bool MirroredDeltas<DeltaTypeSelector<UNDO>>::IsDeltaRelevantForApply(
Timestamp to_include, Timestamp ts) const {
return ts >= to_include;
}
+template<typename T>
+bool MirroredDeltas<T>::IsDeltaRelevantForSelect(
+ Timestamp to_exclude, Timestamp to_include, Timestamp ts) const {
+ return ts >= to_exclude && ts < to_include;
+}
+
+template
+bool MirroredDeltas<DeltaTypeSelector<REDO>>::IsDeltaRelevantForSelect(
+ Timestamp to_exclude, Timestamp to_include, Timestamp ts) const;
+template
+bool MirroredDeltas<DeltaTypeSelector<UNDO>>::IsDeltaRelevantForSelect(
+ Timestamp to_exclude, Timestamp to_include, Timestamp ts) const;
+
} // namespace tablet
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/9f4657ab/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 87daee6..f43e40d 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -365,10 +365,15 @@ class MirroredDeltas {
}
// Returns true if all tracked deltas are irrelevant to 'ts', false otherwise.
- bool CheckAllDeltasCulled(Timestamp ts) const {
+ bool CheckAllDeltasCulled(const boost::optional<Timestamp>& lower_ts,
+ Timestamp upper_ts) const {
for (const auto& e1 : all_deltas_) {
for (const auto& e2 : e1.second) {
- if (IsDeltaRelevant(ts, e2.first)) {
+ bool relevant = IsDeltaRelevantForApply(upper_ts, e2.first);
+ if (lower_ts) {
+ relevant |= IsDeltaRelevantForSelect(*lower_ts, upper_ts, e2.first);
+ }
+ if (relevant) {
return false;
}
}
@@ -401,15 +406,44 @@ class MirroredDeltas {
//
// Rows not set in 'filter' are skipped.
//
- // Deltas not relevant to 'ts' are skipped. The set of rows considered is
- // determined by 'start_row_idx' and the number of rows in 'cb'.
- Status ApplyUpdates(const Schema& projection, Timestamp ts,
+ // Deltas not relevant to 'lower_ts' or 'upper_ts' are skipped. The set of
+ // rows considered is determined by 'start_row_idx' and the number of rows in 'cb'.
+ Status ApplyUpdates(const Schema& projection,
+ const boost::optional<Timestamp>& lower_ts, Timestamp upper_ts,
rowid_t start_row_idx, int col_idx, ColumnBlock* cb,
const SelectionVector& filter) {
+ if (VLOG_IS_ON(3)) {
+ std::string lower_ts_str = lower_ts ? lower_ts->ToString() : "INF";
+ VLOG(3) << "Begin applying for timestamps [" << lower_ts_str << ","
+ << upper_ts << ")";
+ }
for (int i = 0; i < cb->nrows(); i++) {
rowid_t row_idx = start_row_idx + i;
+
+ if (lower_ts) {
+ // First pass: establish whether this row should be applied at all.
+ bool at_least_one_delta_relevant = false;
+ for (const auto& e : all_deltas_[row_idx]) {
+ bool is_relevant = IsDeltaRelevantForSelect(*lower_ts, upper_ts, e.first);
+ if (VLOG_IS_ON(3)) {
+ RowChangeList changes(e.second);
+ VLOG(3) << "Row " << i << " ts " << e.first << " (relevant: "
+ << is_relevant << "): " << changes.ToString(*schema_);
+ }
+ if (is_relevant) {
+ at_least_one_delta_relevant = true;
+ break;
+ }
+ }
+ if (!at_least_one_delta_relevant) {
+ // Not one delta was relevant; skip the row.
+ continue;
+ }
+ }
+
+ // Second pass: apply all relevant deltas.
for (const auto& e : all_deltas_[row_idx]) {
- if (!IsDeltaRelevant(ts, e.first)) {
+ if (!IsDeltaRelevantForApply(upper_ts, e.first)) {
// No need to keep iterating; all future deltas for this row will also
// be irrelevant.
break;
@@ -438,7 +472,7 @@ class MirroredDeltas {
for (int i = 0; i < sel_vec->nrows(); i++) {
bool deleted = false;
for (const auto& e : all_deltas_[start_row_idx + i]) {
- if (!IsDeltaRelevant(ts, e.first)) {
+ if (!IsDeltaRelevantForApply(ts, e.first)) {
// No need to keep iterating; all future deltas for this row will also
// be irrelevant.
break;
@@ -455,6 +489,25 @@ class MirroredDeltas {
return Status::OK();
}
+ // Selects all rows in 'sel_vec' for which there exists a tracked delta.
+ //
+ // Deltas not relevant to 'lower_ts' or 'upper_ts' are skipped. The set of
+ // rows considered is determined by 'start_row_idx' and the number of rows in 'sel_vec'.
+ void SelectUpdates(Timestamp lower_ts, Timestamp upper_ts,
+ rowid_t start_row_idx, SelectionVector* sel_vec) {
+ for (int i = 0; i < sel_vec->nrows(); i++) {
+ for (const auto& e : all_deltas_[start_row_idx + i]) {
+ if (!IsDeltaRelevantForSelect(lower_ts, upper_ts, e.first)) {
+ // Must keep iterating; short-circuit out of the select criteria is
+ // complex and not worth using in test code.
+ continue;
+ }
+ sel_vec->SetRowSelected(i);
+ break;
+ }
+ }
+ }
+
// Transforms and writes deltas into 'deltas', a vector of "delta lists", each
// of which represents a particular row, and each entry of which is a
// Timestamp and encoded delta pair. The encoded delta is a Slice into
@@ -467,7 +520,7 @@ class MirroredDeltas {
std::vector<DeltaList>* deltas) {
for (int i = 0; i < deltas->size(); i++) {
for (const auto& e : all_deltas_[start_row_idx + i]) {
- if (!IsDeltaRelevant(ts, e.first)) {
+ if (!IsDeltaRelevantForApply(ts, e.first)) {
// No need to keep iterating; all future deltas for this row will also
// be irrelevant.
break;
@@ -513,7 +566,13 @@ class MirroredDeltas {
private:
// Returns true if 'ts' is relevant to 'to_include', false otherwise.
- bool IsDeltaRelevant(Timestamp to_include, Timestamp ts) const;
+ bool IsDeltaRelevantForApply(Timestamp to_include, Timestamp ts) const;
+
+ // Returns true if 'ts' is relevant with respect to both 'to_exclude' and
+ // 'to_include', false otherwise.
+ bool IsDeltaRelevantForSelect(Timestamp to_exclude,
+ Timestamp to_include,
+ Timestamp ts) const;
// All encoded deltas, arranged in DeltaKey order.
MirroredDeltaMap all_deltas_;
@@ -779,15 +838,23 @@ void RunDeltaFuzzTest(const DeltaStore& store,
for (int i = 0; i < kNumScans + 1; i++) {
// Pick a timestamp for the iterator. The last iteration will use a snapshot
// that includes all deltas.
- Timestamp ts;
+ Timestamp upper_ts;
+ boost::optional<Timestamp> lower_ts;
if (i < kNumScans) {
- ts = Timestamp(prng->Uniform(ts_range.second - ts_range.first) +
- ts_range.first);
+ uint64_t upper_ts_val = prng->Uniform(ts_range.second - ts_range.first) +
+ ts_range.first;
+ upper_ts = Timestamp(upper_ts_val);
+
+ // Use a lower bound in half the scans.
+ if (prng->Uniform(2)) {
+ uint64_t lower_ts_val = upper_ts_val > 0 ? prng->Uniform(upper_ts_val) : 0;
+ lower_ts = Timestamp(lower_ts_val);
+ }
} else if (T::kTag == REDO) {
- ts = Timestamp::kMax;
+ upper_ts = Timestamp::kMax;
} else {
DCHECK(T::kTag == UNDO);
- ts = Timestamp::kMin;
+ upper_ts = Timestamp::kMin;
}
// Create and initialize the iterator. If none iterator is returned, it's
@@ -796,13 +863,18 @@ void RunDeltaFuzzTest(const DeltaStore& store,
SCOPED_TRACE(strings::Substitute("Projection $0", projection.ToString()));
RowIteratorOptions opts;
opts.projection = &projection;
- opts.snap_to_include = MvccSnapshot(ts);
- SCOPED_TRACE(strings::Substitute("Timestamp $0", ts.ToString()));
+ if (lower_ts) {
+ opts.snap_to_exclude = MvccSnapshot(*lower_ts);
+ }
+ opts.snap_to_include = MvccSnapshot(upper_ts);
+ SCOPED_TRACE(strings::Substitute("Timestamps: [$0,$1)",
+ lower_ts ? lower_ts->ToString() : "INF",
+ upper_ts.ToString()));
DeltaIterator* raw_iter;
Status s = store.NewDeltaIterator(opts, &raw_iter);
if (s.IsNotFound()) {
ASSERT_STR_CONTAINS(s.ToString(), "MvccSnapshot outside the range of this delta");
- ASSERT_TRUE(mirror->CheckAllDeltasCulled(ts));
+ ASSERT_TRUE(mirror->CheckAllDeltasCulled(lower_ts, upper_ts));
continue;
}
ASSERT_OK(s);
@@ -816,20 +888,39 @@ void RunDeltaFuzzTest(const DeltaStore& store,
int batch_size = prng->Uniform(kMaxBatchSize) + 1;
SCOPED_TRACE(strings::Substitute("batch starting at $0 ($1 rows)",
start_row_idx, batch_size));
- ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_APPLY |
- DeltaIterator::PREPARE_FOR_COLLECT));
+ int prepare_flags = DeltaIterator::PREPARE_FOR_APPLY |
+ DeltaIterator::PREPARE_FOR_COLLECT;
+ if (lower_ts) {
+ prepare_flags |= DeltaIterator::PREPARE_FOR_SELECT;
+ }
+ ASSERT_OK(iter->PrepareBatch(batch_size, prepare_flags));
+
+ // Test SelectUpdates: the selection vector begins all false and a row is
+ // set if there is at least one relevant update for it.
+ //
+ // Note: we retain 'actual_selected' for use as a possible filter in the
+ // ApplyUpdates test below.
+ SelectionVector actual_selected(batch_size);
+ if (lower_ts) {
+ SelectionVector expected_selected(batch_size);
+ expected_selected.SetAllFalse();
+ actual_selected.SetAllFalse();
+ mirror->SelectUpdates(*lower_ts, upper_ts, start_row_idx, &expected_selected);
+ ASSERT_OK(iter->SelectUpdates(&actual_selected));
+ ASSERT_EQ(expected_selected, actual_selected);
+ }
// Test ApplyDeletes: the selection vector is all true and a row is unset
// if the last relevant update deleted it.
//
- // Note: we retain 'actual_deleted' for use as a filter in the
+ // Note: we retain 'actual_deleted' for use as a possible filter in the
// ApplyUpdates test below.
SelectionVector actual_deleted(batch_size);
{
SelectionVector expected_deleted(batch_size);
expected_deleted.SetAllTrue();
actual_deleted.SetAllTrue();
- ASSERT_OK(mirror->ApplyDeletes(ts, start_row_idx, &expected_deleted));
+ ASSERT_OK(mirror->ApplyDeletes(upper_ts, start_row_idx, &expected_deleted));
ASSERT_OK(iter->ApplyDeletes(&actual_deleted));
ASSERT_EQ(expected_deleted, actual_deleted);
}
@@ -843,9 +934,10 @@ void RunDeltaFuzzTest(const DeltaStore& store,
expected_scb[k] = 0;
actual_scb[k] = 0;
}
- ASSERT_OK(mirror->ApplyUpdates(*opts.projection, ts, start_row_idx, j,
- &expected_scb, actual_deleted));
- ASSERT_OK(iter->ApplyUpdates(j, &actual_scb, actual_deleted));
+ const SelectionVector& filter = lower_ts ? actual_selected : actual_deleted;
+ ASSERT_OK(mirror->ApplyUpdates(*opts.projection, lower_ts, upper_ts,
+ start_row_idx, j, &expected_scb, filter));
+ ASSERT_OK(iter->ApplyUpdates(j, &actual_scb, filter));
ASSERT_EQ(expected_scb, actual_scb)
<< "Expected column block: " << expected_scb.ToString()
<< "\nActual column block: " << actual_scb.ToString();
@@ -857,7 +949,7 @@ void RunDeltaFuzzTest(const DeltaStore& store,
std::vector<typename MirroredDeltas<T>::DeltaList> expected_muts(batch_size);
std::vector<Mutation*> actual_muts(batch_size);
ASSERT_OK(iter->CollectMutations(&actual_muts, &arena));
- mirror->CollectMutations(ts, start_row_idx, &expected_muts);
+ mirror->CollectMutations(upper_ts, start_row_idx, &expected_muts);
for (int i = 0; i < expected_muts.size(); i++) {
const auto& expected = expected_muts[i];
auto* actual = actual_muts[i];