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 2019/03/06 07:39:05 UTC
[kudu] branch master updated (e2c611b -> cd66342)
This is an automated email from the ASF dual-hosted git repository.
mpercy pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from e2c611b [docs] Remove outdated package and quickstart references
new 5953357 KUDU-2645: tablet: Support deduplication of deleted rows in MergeIterator
new cd66342 java: ITClient: name background threads
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../test/java/org/apache/kudu/client/ITClient.java | 17 +--
.../java/org/apache/kudu/test/TempDirUtils.java | 1 +
src/kudu/common/generic_iterators-test.cc | 146 +++++++++++++++----
src/kudu/common/generic_iterators.cc | 160 ++++++++++++++++-----
src/kudu/common/generic_iterators.h | 12 ++
src/kudu/common/schema.cc | 4 +-
src/kudu/common/schema.h | 2 +-
src/kudu/tablet/diff_scan-test.cc | 61 ++++++--
src/kudu/tablet/rowset.cc | 2 +-
src/kudu/tablet/tablet.cc | 2 +-
10 files changed, 315 insertions(+), 92 deletions(-)
[kudu] 02/02: java: ITClient: name background threads
Posted by mp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mpercy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit cd66342f581307522b3718237db15b252ff59a3e
Author: Mike Percy <mp...@apache.org>
AuthorDate: Tue Mar 5 19:02:20 2019 -0800
java: ITClient: name background threads
The ITClient test has several background threads that perform tasks
while the test is running. Naming these threads makes it easier to
understand what happened when wading through the test log.
This patch also squelches an unrelated warning from TempDirUtils due to
not specifying @Override on a Thread's run() method.
Change-Id: I9874b088cc8e77707fc913df3de1d1592c1de6c3
Reviewed-on: http://gerrit.cloudera.org:8080/12677
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
---
.../src/test/java/org/apache/kudu/client/ITClient.java | 17 +++++------------
.../main/java/org/apache/kudu/test/TempDirUtils.java | 1 +
2 files changed, 6 insertions(+), 12 deletions(-)
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
index fa85efc..5998a27 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/ITClient.java
@@ -92,20 +92,13 @@ public class ITClient {
@Test(timeout = TEST_TIMEOUT_SECONDS)
public void test() throws Exception {
-
- UncaughtExceptionHandler uncaughtExceptionHandler = new UncaughtExceptionHandler();
-
- ArrayList<Thread> threads = new ArrayList<>();
- Thread chaosThread = new Thread(new ChaosThread());
- Thread writerThread = new Thread(new WriterThread());
- Thread scannerThread = new Thread(new ScannerThread());
-
- threads.add(chaosThread);
- threads.add(writerThread);
- threads.add(scannerThread);
+ List<Thread> threads = new ArrayList<>();
+ threads.add(new Thread(new ChaosThread(), "chaos-test-thread"));
+ threads.add(new Thread(new WriterThread(), "writer-test-thread"));
+ threads.add(new Thread(new ScannerThread(), "scanner-test-thread"));
for (Thread thread : threads) {
- thread.setUncaughtExceptionHandler(uncaughtExceptionHandler);
+ thread.setUncaughtExceptionHandler(new UncaughtExceptionHandler());
thread.start();
}
diff --git a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/TempDirUtils.java b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/TempDirUtils.java
index 8fa65a2..75d5add 100644
--- a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/TempDirUtils.java
+++ b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/TempDirUtils.java
@@ -86,6 +86,7 @@ public class TempDirUtils {
private static void registerToRecursivelyDeleteOnShutdown(Path path) {
final Path absPath = path.toAbsolutePath();
Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
public void run() {
File dir = absPath.toFile();
if (!dir.exists()) return;
[kudu] 01/02: KUDU-2645: tablet: Support deduplication of deleted
rows in MergeIterator
Posted by mp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mpercy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 5953357c486062781ac04db7e61fea0ea0a3cc07
Author: Mike Percy <mp...@apache.org>
AuthorDate: Mon Jan 28 12:49:02 2019 -0800
KUDU-2645: tablet: Support deduplication of deleted rows in MergeIterator
This patch makes it possible to do an incremental diff scan of an entire
tablet. A follow-up patch will expose this capability to scanners at the
RPC level.
Included is new support in the MergeIterator for deduplicating deleted
rows resulting from when a row with a particular primary key is deleted
from one rowset and reinserted into a different rowset. These duplicates
may be returned when the 'include_deleted_rows' option is enabled in the
row iterator options.
Because of this new behavior, while the MergeIterator will deduplicate
deleted rows if they are included in the result set, the UnionIterator
will not deduplicate them and instead return all instances found,
because it's not possible to efficiently support such deduplication
without a merge-like process.
One tangentially-related change in this patch is that TestMerge in
generic_iterators-test.cc was modified to no longer generate duplicate
non-deleted keys for merge testing. Duplicate non-deleted row keys are
no longer supported in the MergeIterator since there is currently no
practical use for that, and it's more efficient not to support them
since at the time of writing it isn't possible for them to appear in a
real tablet.
This patch adds tests at a couple of different levels, including a
VectorIterator-based test that is useful for benchmarking MergeIterator
performance on a simple schema, as well as a higher-level diff scan
test that operates at the Tablet level.
Change-Id: I00614b3fa5c6b4e7b620bb78489e24c5ad44daee
Reviewed-on: http://gerrit.cloudera.org:8080/12205
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Mike Percy <mp...@apache.org>
---
src/kudu/common/generic_iterators-test.cc | 146 +++++++++++++++++++++------
src/kudu/common/generic_iterators.cc | 160 ++++++++++++++++++++++++------
src/kudu/common/generic_iterators.h | 12 +++
src/kudu/common/schema.cc | 4 +-
src/kudu/common/schema.h | 2 +-
src/kudu/tablet/diff_scan-test.cc | 61 +++++++++---
src/kudu/tablet/rowset.cc | 2 +-
src/kudu/tablet/tablet.cc | 2 +-
8 files changed, 309 insertions(+), 80 deletions(-)
diff --git a/src/kudu/common/generic_iterators-test.cc b/src/kudu/common/generic_iterators-test.cc
index 375be34..2578491 100644
--- a/src/kudu/common/generic_iterators-test.cc
+++ b/src/kudu/common/generic_iterators-test.cc
@@ -25,6 +25,7 @@
#include <random>
#include <string>
#include <unordered_map>
+#include <unordered_set>
#include <vector>
#include <gflags/gflags.h>
@@ -42,6 +43,7 @@
#include "kudu/common/scan_spec.h"
#include "kudu/common/schema.h"
#include "kudu/common/types.h"
+#include "kudu/gutil/map-util.h"
#include "kudu/gutil/mathlimits.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/memory/arena.h"
@@ -58,6 +60,7 @@ DECLARE_bool(materializing_iterator_do_pushdown);
using std::string;
using std::unique_ptr;
+using std::unordered_set;
using std::vector;
using strings::Substitute;
@@ -65,17 +68,32 @@ namespace kudu {
struct IteratorStats;
-static const Schema kIntSchema({ ColumnSchema("val", UINT32) }, 1);
+static const int kValColIdx = 0; // Index of 'val' column in these test schemas.
+static const Schema kIntSchema({ ColumnSchema("val", UINT32) },
+ /*key_columns=*/1);
+static const bool kIsDeletedReadDefault = false;
+static const Schema kIntSchemaWithVCol({ ColumnSchema("val", UINT32),
+ ColumnSchema("is_deleted", IS_DELETED,
+ /*is_nullable=*/false,
+ /*read_default=*/&kIsDeletedReadDefault) },
+ /*key_columns=*/1);
// Test iterator which just yields integer rows from a provided
// vector.
class VectorIterator : public ColumnwiseIterator {
public:
- explicit VectorIterator(vector<uint32_t> ints)
+ VectorIterator(vector<uint32_t> ints, vector<uint8_t> is_deleted, Schema schema)
: ints_(std::move(ints)),
+ is_deleted_(std::move(is_deleted)),
+ schema_(std::move(schema)),
cur_idx_(0),
block_size_(ints_.size()),
sel_vec_(nullptr) {
+ CHECK_EQ(ints_.size(), is_deleted_.size());
+ }
+
+ explicit VectorIterator(const vector<uint32_t>& ints)
+ : VectorIterator(ints, vector<uint8_t>(ints.size()), kIntSchema) {
}
// Set the number of rows that will be returned in each
@@ -120,11 +138,21 @@ class VectorIterator : public ColumnwiseIterator {
Status MaterializeColumn(ColumnMaterializationContext* ctx) override {
ctx->SetDecoderEvalNotSupported();
- CHECK_EQ(UINT32, ctx->block()->type_info()->physical_type());
DCHECK_LE(prepared_, ctx->block()->nrows());
- for (size_t i = 0; i < prepared_; i++) {
- ctx->block()->SetCellValue(i, &(ints_[cur_idx_ + i]));
+ switch (ctx->block()->type_info()->physical_type()) {
+ case UINT32:
+ for (size_t i = 0; i < prepared_; i++) {
+ ctx->block()->SetCellValue(i, &(ints_[cur_idx_ + i]));
+ }
+ break;
+ case BOOL:
+ for (size_t i = 0; i < prepared_; i++) {
+ ctx->block()->SetCellValue(i, &(is_deleted_[cur_idx_ + i]));
+ }
+ break;
+ default:
+ LOG(FATAL) << "unsupported column type in VectorIterator";
}
return Status::OK();
@@ -144,8 +172,8 @@ class VectorIterator : public ColumnwiseIterator {
return Substitute("VectorIterator [$0,$1]", ints_[0], ints_[ints_.size() - 1]);
}
- const Schema &schema() const override {
- return kIntSchema;
+ const Schema& schema() const override {
+ return schema_;
}
void GetIteratorStats(vector<IteratorStats>* stats) const override {
@@ -154,6 +182,11 @@ class VectorIterator : public ColumnwiseIterator {
private:
vector<uint32_t> ints_;
+ // We use vector<uint8_t> instead of vector<bool> to represent the IS_DELETED
+ // column so we can call ColumnBlock::SetCellValue() in MaterializeColumn(),
+ // whose API requires taking an address to a non-temporary for the value.
+ vector<uint8_t> is_deleted_;
+ const Schema schema_;
int cur_idx_;
int block_size_;
size_t prepared_;
@@ -167,7 +200,8 @@ TEST(TestMergeIterator, TestMergeEmpty) {
unique_ptr<ColumnwiseIterator>(new VectorIterator({}))));
vector<unique_ptr<RowwiseIterator>> input;
input.emplace_back(std::move(iter));
- unique_ptr<RowwiseIterator> merger(NewMergeIterator(std::move(input)));
+ unique_ptr<RowwiseIterator> merger(
+ NewMergeIterator(MergeIteratorOptions(/*include_deleted_rows=*/false), std::move(input)));
ASSERT_OK(merger->Init(nullptr));
ASSERT_FALSE(merger->HasNext());
}
@@ -182,44 +216,82 @@ TEST(TestMergeIterator, TestMergeEmptyViaSelectionVector) {
unique_ptr<RowwiseIterator> iter(NewMaterializingIterator(std::move(vec)));
vector<unique_ptr<RowwiseIterator>> input;
input.emplace_back(std::move(iter));
- unique_ptr<RowwiseIterator> merger(NewMergeIterator(std::move(input)));
+ unique_ptr<RowwiseIterator> merger(
+ NewMergeIterator(MergeIteratorOptions(/*include_deleted_rows=*/false), std::move(input)));
ASSERT_OK(merger->Init(nullptr));
ASSERT_FALSE(merger->HasNext());
}
class TestIntRangePredicate {
public:
- TestIntRangePredicate(uint32_t lower, uint32_t upper) :
- lower_(lower),
- upper_(upper),
- pred_(ColumnPredicate::Range(kIntSchema.column(0), &lower_, &upper_)) {}
+ TestIntRangePredicate(uint32_t lower, uint32_t upper, const ColumnSchema& column)
+ : lower_(lower),
+ upper_(upper),
+ pred_(ColumnPredicate::Range(column, &lower_, &upper_)) {
+ }
+
+ TestIntRangePredicate(uint32_t lower, uint32_t upper)
+ : TestIntRangePredicate(lower, upper, kIntSchema.column(0)) {
+ }
uint32_t lower_, upper_;
ColumnPredicate pred_;
};
-void TestMerge(const TestIntRangePredicate &predicate, bool overlapping_inputs = true) {
+void TestMerge(const Schema& schema, const TestIntRangePredicate &predicate,
+ bool overlapping_ranges = true, bool include_deleted_rows = false) {
struct List {
vector<uint32_t> ints;
+ vector<uint8_t> is_deleted;
unique_ptr<SelectionVector> sv;
};
vector<List> all_ints;
vector<uint32_t> expected;
+ unordered_set<uint32_t> expected_ints_deleted;
+ unordered_set<uint32_t> seen_live;
expected.reserve(FLAGS_num_rows * FLAGS_num_lists);
Random prng(SeedRandom());
uint32_t entry = 0;
for (int i = 0; i < FLAGS_num_lists; i++) {
vector<uint32_t> ints;
+ vector<uint8_t> deleted_status;
ints.reserve(FLAGS_num_rows);
unique_ptr<SelectionVector> sv(new SelectionVector(FLAGS_num_rows));
- if (overlapping_inputs) {
+ if (overlapping_ranges) {
entry = 0;
}
for (int j = 0; j < FLAGS_num_rows; j++) {
- entry += prng.Uniform(5);
+ uint32_t potential;
+ bool is_deleted = false;
+ // The merge iterator does not support duplicate non-deleted keys.
+ while (true) {
+ potential = entry + prng.Uniform(FLAGS_num_rows * FLAGS_num_lists * 10);
+ if (ContainsKey(seen_live, potential)) continue;
+
+ // If we are including deleted rows, with some probability make this a
+ // deleted row.
+ if (include_deleted_rows) {
+ is_deleted = prng.OneIn(4);
+ if (is_deleted) {
+ // Including several duplicate deleted keys is fine.
+ expected_ints_deleted.insert(potential);
+ break;
+ }
+ }
+
+ // This is a new live row. Un-mark it as deleted if necessary.
+ InsertOrDie(&seen_live, potential);
+ if (include_deleted_rows && ContainsKey(expected_ints_deleted, potential)) {
+ CHECK_EQ(1, expected_ints_deleted.erase(potential));
+ }
+ break;
+ }
+ entry = potential;
+
ints.emplace_back(entry);
+ deleted_status.emplace_back(is_deleted);
// Some entries are randomly deselected in order to exercise the selection
// vector logic in the MergeIterator. This is reflected both in the input
@@ -239,7 +311,7 @@ void TestMerge(const TestIntRangePredicate &predicate, bool overlapping_inputs =
}
}
- all_ints.emplace_back(List{ std::move(ints), std::move(sv) });
+ all_ints.emplace_back(List{ std::move(ints), std::move(deleted_status), std::move(sv) });
}
LOG_TIMING(INFO, "sorting the expected results") {
@@ -254,10 +326,11 @@ void TestMerge(const TestIntRangePredicate &predicate, bool overlapping_inputs =
VLOG(1) << "Predicate expects " << expected.size() << " results: " << expected;
+ const int kIsDeletedIndex = schema.find_first_is_deleted_virtual_column();
for (int trial = 0; trial < FLAGS_num_iters; trial++) {
vector<unique_ptr<RowwiseIterator>> to_merge;
for (const auto& e : all_ints) {
- unique_ptr<VectorIterator> vec_it(new VectorIterator(e.ints));
+ unique_ptr<VectorIterator> vec_it(new VectorIterator(e.ints, e.is_deleted, schema));
vec_it->set_block_size(10);
vec_it->set_selection_vector(e.sv.get());
unique_ptr<RowwiseIterator> mat_it(NewMaterializingIterator(std::move(vec_it)));
@@ -273,10 +346,12 @@ void TestMerge(const TestIntRangePredicate &predicate, bool overlapping_inputs =
LOG(INFO) << "Predicate: " << predicate.pred_.ToString();
LOG_TIMING(INFO, "iterating merged lists") {
- unique_ptr<RowwiseIterator> merger(NewMergeIterator({ std::move(to_merge) }));
+ unique_ptr<RowwiseIterator> merger(
+ NewMergeIterator(MergeIteratorOptions(include_deleted_rows),
+ std::move(to_merge)));
ASSERT_OK(merger->Init(&spec));
- RowBlock dst(kIntSchema, 100, nullptr);
+ RowBlock dst(schema, 100, nullptr);
size_t total_idx = 0;
while (merger->HasNext()) {
ASSERT_OK(merger->NextBlock(&dst));
@@ -284,13 +359,19 @@ void TestMerge(const TestIntRangePredicate &predicate, bool overlapping_inputs =
"if HasNext() returns true, must return some rows";
for (int i = 0; i < dst.nrows(); i++) {
- uint32_t this_row = *kIntSchema.ExtractColumnFromRow<UINT32>(dst.row(i), 0);
- ASSERT_GE(this_row, predicate.lower_) << "Yielded integer excluded by predicate";
- ASSERT_LT(this_row, predicate.upper_) << "Yielded integer excluded by predicate";
- if (expected[total_idx] != this_row) {
- ASSERT_EQ(expected[total_idx], this_row) <<
+ uint32_t row_val = *schema.ExtractColumnFromRow<UINT32>(dst.row(i), kValColIdx);
+ ASSERT_GE(row_val, predicate.lower_) << "Yielded integer excluded by predicate";
+ ASSERT_LT(row_val, predicate.upper_) << "Yielded integer excluded by predicate";
+ if (expected[total_idx] != row_val) {
+ ASSERT_EQ(expected[total_idx], row_val) <<
"Yielded out of order at idx " << total_idx;
}
+ if (include_deleted_rows) {
+ ASSERT_NE(Schema::kColumnNotFound, kIsDeletedIndex);
+ bool is_deleted = *schema.ExtractColumnFromRow<IS_DELETED>(dst.row(i), kIsDeletedIndex);
+ ASSERT_EQ(is_deleted, ContainsKey(expected_ints_deleted, row_val))
+ << "Row " << row_val << " has unexpected IS_DELETED value at index " << total_idx;
+ }
total_idx++;
}
}
@@ -304,17 +385,17 @@ void TestMerge(const TestIntRangePredicate &predicate, bool overlapping_inputs =
TEST(TestMergeIterator, TestMerge) {
TestIntRangePredicate predicate(0, MathLimits<uint32_t>::kMax);
- TestMerge(predicate);
+ NO_FATALS(TestMerge(kIntSchema, predicate));
}
TEST(TestMergeIterator, TestMergeNonOverlapping) {
TestIntRangePredicate predicate(0, MathLimits<uint32_t>::kMax);
- TestMerge(predicate, /*overlapping_inputs=*/false);
+ NO_FATALS(TestMerge(kIntSchema, predicate, /*overlapping_ranges=*/false));
}
TEST(TestMergeIterator, TestMergePredicate) {
TestIntRangePredicate predicate(0, FLAGS_num_rows / 5);
- TestMerge(predicate);
+ NO_FATALS(TestMerge(kIntSchema, predicate));
}
// Regression test for a bug in the merge which would incorrectly
@@ -323,7 +404,14 @@ TEST(TestMergeIterator, TestMergePredicate) {
// second half.
TEST(TestMergeIterator, TestMergePredicate2) {
TestIntRangePredicate predicate(FLAGS_num_rows / 2, MathLimits<uint32_t>::kMax);
- TestMerge(predicate);
+ NO_FATALS(TestMerge(kIntSchema, predicate));
+}
+
+TEST(TestMergeIterator, TestDeDupGhostRows) {
+ TestIntRangePredicate match_all_pred(0, MathLimits<uint32_t>::kMax);
+ NO_FATALS(TestMerge(kIntSchemaWithVCol, match_all_pred,
+ /*overlapping_ranges=*/true,
+ /*include_deleted_rows=*/true));
}
// Test that the MaterializingIterator properly evaluates predicates when they apply
diff --git a/src/kudu/common/generic_iterators.cc b/src/kudu/common/generic_iterators.cc
index 58b23d5..682c2e1 100644
--- a/src/kudu/common/generic_iterators.cc
+++ b/src/kudu/common/generic_iterators.cc
@@ -17,8 +17,6 @@
#include "kudu/common/generic_iterators.h"
-#include <unistd.h>
-
#include <algorithm>
#include <cstddef>
#include <cstdint>
@@ -29,6 +27,7 @@
#include <ostream>
#include <string>
#include <unordered_map>
+#include <unordered_set>
#include <utility>
#include <gflags/gflags.h>
@@ -37,6 +36,7 @@
#include "kudu/common/column_materialization_context.h"
#include "kudu/common/column_predicate.h"
#include "kudu/common/columnblock.h"
+#include "kudu/common/common.pb.h"
#include "kudu/common/iterator.h"
#include "kudu/common/iterator_stats.h"
#include "kudu/common/row.h"
@@ -63,6 +63,7 @@ using std::shared_ptr;
using std::sort;
using std::string;
using std::unique_ptr;
+using std::unordered_set;
using std::vector;
using strings::Substitute;
@@ -235,7 +236,7 @@ class MergeIterator : public RowwiseIterator {
//
// Note: the iterators must be constructed using a projection that includes
// all key columns; otherwise a CHECK will fire at initialization time.
- explicit MergeIterator(vector<unique_ptr<RowwiseIterator>> iters);
+ MergeIterator(MergeIteratorOptions opts, vector<unique_ptr<RowwiseIterator>> iters);
virtual ~MergeIterator();
@@ -259,11 +260,17 @@ class MergeIterator : public RowwiseIterator {
Status MaterializeBlock(RowBlock* dst);
Status InitSubIterators(ScanSpec *spec);
+ const MergeIteratorOptions opts_;
+
// Initialized during Init.
unique_ptr<Schema> schema_;
bool initted_;
+ // Index of the IS_DELETED column, or Schema::kColumnNotFound if no such
+ // column exists in the schema.
+ int is_deleted_col_index_;
+
// Holds the subiterators until Init is called, at which point this is cleared.
// This is required because we can't create a MergeIterState of an uninitialized iterator.
vector<unique_ptr<RowwiseIterator>> orig_iters_;
@@ -290,8 +297,10 @@ class MergeIterator : public RowwiseIterator {
int64_t num_comparisons_;
};
-MergeIterator::MergeIterator(vector<unique_ptr<RowwiseIterator>> iters)
- : initted_(false),
+MergeIterator::MergeIterator(MergeIteratorOptions opts,
+ vector<unique_ptr<RowwiseIterator>> iters)
+ : opts_(opts),
+ initted_(false),
orig_iters_(std::move(iters)),
num_orig_iters_(orig_iters_.size()),
num_comparisons_(0) {
@@ -335,6 +344,12 @@ Status MergeIterator::Init(ScanSpec *spec) {
}
#endif
+ is_deleted_col_index_ = schema_->find_first_is_deleted_virtual_column();
+ if (opts_.include_deleted_rows && is_deleted_col_index_ == Schema::kColumnNotFound) {
+ return Status::InvalidArgument("Merge iterator cannot deduplicate deleted "
+ "rows without an IS_DELETED column");
+ }
+
// Before we copy any rows, clean up any iterators which were empty
// to start with. Otherwise, HasNext() won't properly return false
// if we were passed only empty iterators.
@@ -358,7 +373,7 @@ Status MergeIterator::InitSubIterators(ScanSpec *spec) {
for (auto& i : orig_iters_) {
ScanSpec *spec_copy = spec != nullptr ? scan_spec_copies_.Construct(*spec) : nullptr;
RETURN_NOT_OK(InitAndMaybeWrap(&i, spec_copy));
- states_.push_back(unique_ptr<MergeIterState>(new MergeIterState(std::move(i))));
+ states_.emplace_back(unique_ptr<MergeIterState>(new MergeIterState(std::move(i))));
}
orig_iters_.clear();
@@ -399,47 +414,126 @@ void MergeIterator::PrepareBatch(RowBlock* dst) {
// and such around the comparisons. A simple experiment indicated there's some
// 2x to be gained.
Status MergeIterator::MaterializeBlock(RowBlock *dst) {
+ // We need a vector to track the iterators whose next_row() contains the
+ // smallest row key at a given moment during the merge because there may be
+ // multiple deleted rows with the same row key across multiple rowsets, and
+ // up to one live instance, that we have to deduplicate.
+ vector<MergeIterState*> smallest(states_.size());
+
// Initialize the selection vector.
// MergeIterState only returns selected rows.
dst->selection_vector()->SetAllTrue();
- for (size_t dst_row_idx = 0; dst_row_idx < dst->nrows(); dst_row_idx++) {
- RowBlockRow dst_row = dst->row(dst_row_idx);
+ size_t dst_row_idx = 0;
+ while (dst_row_idx < dst->nrows()) {
- // Find the sub-iterator which is currently smallest
- MergeIterState *smallest = nullptr;
- ssize_t smallest_idx = -1;
+ // Find the sub-iterator that is currently smallest.
+ smallest.clear();
// Typically the number of states_ is not that large, so using a priority
- // queue is not worth it
- for (size_t i = 0; i < states_.size(); i++) {
- unique_ptr<MergeIterState> &state = states_[i];
-
- if (PREDICT_FALSE(smallest == nullptr)) {
- smallest = state.get();
- smallest_idx = i;
- } else {
+ // queue is not worth it.
+ for (const auto& iter : states_) {
+ // To merge in row key order, we need to consume the smallest row at any
+ // given time. We locate that row by peeking at the next row in each of
+ // the states_ iterators, which includes all possible candidates for the
+ // next row in key order.
+ int cmp;
+ if (!smallest.empty()) {
+ // If we have a candidate for smallest row, compare it against the
+ // smallest row in each iterator.
+ cmp = schema_->Compare(iter->next_row(), smallest[0]->next_row());
num_comparisons_++;
- if (schema_->Compare(state->next_row(), smallest->next_row()) < 0) {
- smallest = state.get();
- smallest_idx = i;
- }
+ }
+ if (smallest.empty() || cmp < 0) {
+ // If we have no candidates for the next row yet, or the row found is
+ // smaller than the previously-smallest, replace the smallest with the
+ // new row found.
+ smallest.clear();
+ smallest.emplace_back(iter.get());
+ } else if (!smallest.empty() && cmp == 0) {
+ // If we have found a duplicate of the smallest row, at least one must
+ // be a ghost row. Collect all duplicates in order to merge them later.
+ smallest.emplace_back(iter.get());
}
}
// If no iterators had any row left, then we're done iterating.
- if (PREDICT_FALSE(smallest == nullptr)) break;
-
- // Otherwise, copy the row from the smallest one, and advance it
- RETURN_NOT_OK(CopyRow(smallest->next_row(), &dst_row, dst->arena()));
- RETURN_NOT_OK(smallest->Advance());
+ if (PREDICT_FALSE(smallest.empty())) break;
+
+ MergeIterState* row_to_return_iter = nullptr;
+ if (!opts_.include_deleted_rows) {
+ // Since deleted rows are not included here, there can only be a single
+ // instance of any given row key in 'smallest'.
+ CHECK_EQ(1, smallest.size()) << "expected only a single smallest row";
+ row_to_return_iter = smallest[0];
+ } else {
+ // Deduplicate any deleted rows. Row instance de-duplication criteria:
+ // 1. If there is a non-deleted instance, return that instance.
+ // 2. If all rows are deleted, any instance will suffice because we
+ // don't guarantee that we will return valid field values for deleted
+ // rows.
+ int live_rows_found = 0;
+ for (const auto& s : smallest) {
+ bool is_deleted =
+ *schema_->ExtractColumnFromRow<IS_DELETED>(s->next_row(), is_deleted_col_index_);
+ if (!is_deleted) {
+ // We found the single live instance of the row.
+ row_to_return_iter = s;
+#ifndef NDEBUG
+ live_rows_found++; // In DEBUG mode, do a sanity-check count of the live rows.
+#else
+ break; // In RELEASE mode, short-circuit the loop.
+#endif
+ }
+ }
+ DCHECK_LE(live_rows_found, 1) << "expected at most one live row";
+
+ // If all instances of a given row are deleted then return an arbitrary
+ // deleted instance.
+ if (row_to_return_iter == nullptr) {
+ row_to_return_iter = smallest[0];
+ DCHECK(*schema_->ExtractColumnFromRow<IS_DELETED>(row_to_return_iter->next_row(),
+ is_deleted_col_index_))
+ << "expected deleted row";
+ }
+ }
+ RowBlockRow dst_row = dst->row(dst_row_idx++);
+ RETURN_NOT_OK(CopyRow(row_to_return_iter->next_row(), &dst_row, dst->arena()));
+
+ // Advance all matching sub-iterators and mark exhausted sub-iterators for
+ // removal.
+ unordered_set<MergeIterState*> exhausted;
+ for (MergeIterState* s : smallest) {
+ RETURN_NOT_OK(s->Advance());
+ if (s->IsFullyExhausted()) {
+ InsertOrDie(&exhausted, s);
+ }
+ }
- if (smallest->IsFullyExhausted()) {
+ // Remove the exhausted sub-iterators.
+ if (!exhausted.empty()) {
std::lock_guard<rw_spinlock> l(states_lock_);
- smallest->AddStats(&finished_iter_stats_by_col_);
- states_.erase(states_.begin() + smallest_idx);
+ for (MergeIterState* s : exhausted) {
+ s->AddStats(&finished_iter_stats_by_col_);
+ }
+ // TODO(mpercy): Consider making removal O(1) per element to remove by
+ // using a different data structure for 'states_'. The below
+ // erase-remove idiom gives us O(n) removal on a vector for an
+ // arbitrary number of elements to remove at once.
+ states_.erase(std::remove_if(states_.begin(), states_.end(),
+ [&exhausted](const unique_ptr<MergeIterState>& state) {
+ return ContainsKey(exhausted, state.get());
+ }),
+ states_.end());
}
}
+ // The number of rows actually copied to the destination RowBlock may be less
+ // than its original capacity due to deduplication of ghost rows.
+ DCHECK_LE(dst_row_idx, dst->nrows());
+ if (dst_row_idx < dst->nrows()) {
+ dst->Resize(dst_row_idx);
+ }
+
return Status::OK();
}
@@ -463,8 +557,8 @@ void MergeIterator::GetIteratorStats(vector<IteratorStats>* stats) const {
}
unique_ptr<RowwiseIterator> NewMergeIterator(
- vector<unique_ptr<RowwiseIterator>> iters) {
- return unique_ptr<RowwiseIterator>(new MergeIterator(std::move(iters)));
+ MergeIteratorOptions opts, vector<unique_ptr<RowwiseIterator>> iters) {
+ return unique_ptr<RowwiseIterator>(new MergeIterator(opts, std::move(iters)));
}
int64_t GetMergeIteratorNumComparisonsForTests(
diff --git a/src/kudu/common/generic_iterators.h b/src/kudu/common/generic_iterators.h
index 84ab636..2c3fccf 100644
--- a/src/kudu/common/generic_iterators.h
+++ b/src/kudu/common/generic_iterators.h
@@ -29,10 +29,22 @@ class ColumnwiseIterator;
class RowwiseIterator;
class ScanSpec;
+// Options struct for the MergeIterator.
+struct MergeIteratorOptions {
+ explicit MergeIteratorOptions(bool include_deleted_rows)
+ : include_deleted_rows(include_deleted_rows) {}
+
+ // Whether to include and de-duplicate ghost (deleted) rows. If true, the
+ // sub-iterator schemas must also contain an IS_DELETED column or an error
+ // will be returned during MergeIterator initialization.
+ const bool include_deleted_rows;
+};
+
// Constructs a MergeIterator of the given iterators.
//
// The iterators must have matching schemas and should not yet be initialized.
std::unique_ptr<RowwiseIterator> NewMergeIterator(
+ MergeIteratorOptions opts,
std::vector<std::unique_ptr<RowwiseIterator>> iters);
// Constructs a UnionIterator of the given iterators.
diff --git a/src/kudu/common/schema.cc b/src/kudu/common/schema.cc
index 3e4132e..c5ca4f4 100644
--- a/src/kudu/common/schema.cc
+++ b/src/kudu/common/schema.cc
@@ -150,9 +150,11 @@ size_t ColumnSchema::memory_footprint_including_this() const {
return kudu_malloc_usable_size(this) + memory_footprint_excluding_this();
}
+const int Schema::kColumnNotFound = -1;
+
Schema::Schema(const Schema& other)
: name_to_index_bytes_(0),
- name_to_index_(/*bucket_count=*/10,
+ name_to_index_(/*bucket_count*/ 10,
NameToIndexMap::hasher(),
NameToIndexMap::key_equal(),
NameToIndexMapAllocator(&name_to_index_bytes_)) {
diff --git a/src/kudu/common/schema.h b/src/kudu/common/schema.h
index 322e87c..9886a61 100644
--- a/src/kudu/common/schema.h
+++ b/src/kudu/common/schema.h
@@ -420,7 +420,7 @@ class ColumnSchema {
class Schema {
public:
- static const int kColumnNotFound = -1;
+ static const int kColumnNotFound;
Schema()
: num_key_columns_(0),
diff --git a/src/kudu/tablet/diff_scan-test.cc b/src/kudu/tablet/diff_scan-test.cc
index bd4ae26..ba7aa0f 100644
--- a/src/kudu/tablet/diff_scan-test.cc
+++ b/src/kudu/tablet/diff_scan-test.cc
@@ -15,10 +15,10 @@
// specific language governing permissions and limitations
// under the License.
+#include <algorithm>
#include <cstdint>
#include <memory>
#include <string>
-#include <utility>
#include <vector>
#include <gtest/gtest.h>
@@ -34,6 +34,7 @@
#include "kudu/tablet/tablet-test-base.h"
#include "kudu/tablet/tablet-test-util.h"
#include "kudu/tablet/tablet.h"
+#include "kudu/tablet/tablet_metadata.h"
#include "kudu/util/test_macros.h"
using std::string;
@@ -43,7 +44,8 @@ using std::vector;
namespace kudu {
namespace tablet {
-class DiffScanTest : public TabletTestBase<IntKeyTestSetup<INT64>> {
+class DiffScanTest : public TabletTestBase<IntKeyTestSetup<INT64>>,
+ public ::testing::WithParamInterface<std::tuple<OrderMode, bool>> {
public:
DiffScanTest()
: Superclass(TabletHarness::Options::ClockType::HYBRID_CLOCK) {}
@@ -52,7 +54,14 @@ class DiffScanTest : public TabletTestBase<IntKeyTestSetup<INT64>> {
using Superclass = TabletTestBase<IntKeyTestSetup<INT64>>;
};
-TEST_F(DiffScanTest, TestDiffScan) {
+INSTANTIATE_TEST_CASE_P(DiffScanModes, DiffScanTest,
+ ::testing::Combine(
+ /*order_mode*/ ::testing::Values(UNORDERED, ORDERED),
+ /*include_deleted_rows*/ ::testing::Bool()));
+
+TEST_P(DiffScanTest, TestDiffScan) {
+ OrderMode order_mode = std::get<0>(GetParam());
+ bool include_deleted_rows = std::get<1>(GetParam());
auto tablet = this->tablet();
auto tablet_id = tablet->tablet_id();
@@ -85,10 +94,20 @@ TEST_F(DiffScanTest, TestDiffScan) {
RowIteratorOptions opts;
opts.snap_to_include = snap2;
- opts.order = ORDERED;
- opts.include_deleted_rows = true;
-
- auto projection = tablet->schema()->CopyWithoutColumnIds();
+ opts.order = order_mode;
+ opts.include_deleted_rows = include_deleted_rows;
+
+ static const bool kIsDeletedDefault = false;
+ SchemaBuilder builder(tablet->metadata()->schema());
+ if (order_mode == ORDERED) {
+ // The merge iterator requires an IS_DELETED column when including deleted
+ // rows in order to support deduplication of the rows.
+ ASSERT_OK(builder.AddColumn("deleted", IS_DELETED,
+ /*is_nullable=*/ false,
+ /*read_default=*/ &kIsDeletedDefault,
+ /*write_default=*/ nullptr));
+ }
+ Schema projection = builder.BuildWithoutIds();
opts.projection = &projection;
unique_ptr<RowwiseIterator> row_iterator;
@@ -98,14 +117,28 @@ TEST_F(DiffScanTest, TestDiffScan) {
ScanSpec spec;
ASSERT_OK(row_iterator->Init(&spec));
- // For the time being, we should get two rows back, and they should both be
- // the same key. In reality, one has been deleted.
- // TODO(KUDU-2645): The result of this test should change once we properly
- // implement diff scans and the merge iterator is able to deduplicate ghosts.
ASSERT_OK(tablet::IterateToStringList(row_iterator.get(), &rows));
- ASSERT_EQ(2, rows.size());
- EXPECT_EQ("(int64 key=1, int32 key_idx=1, int32 val=1)", rows[0]);
- EXPECT_EQ("(int64 key=1, int32 key_idx=1, int32 val=2)", rows[1]);
+
+ // In unordered mode, the union iterator will not deduplicate row keys.
+ // In ordered mode, the merge iterator will perform deduplication.
+ if (order_mode == UNORDERED) {
+ if (include_deleted_rows) {
+ // No de-dup.
+ ASSERT_EQ(2, rows.size());
+ // There is no guaranteed order of these results so get them in alpha order.
+ std::sort(rows.begin(), rows.end());
+ EXPECT_EQ("(int64 key=1, int32 key_idx=1, int32 val=1)", rows[0]);
+ EXPECT_EQ("(int64 key=1, int32 key_idx=1, int32 val=2)", rows[1]);
+ } else {
+ // There will only ever be a single live version of any one row.
+ ASSERT_EQ(1, rows.size());
+ EXPECT_EQ("(int64 key=1, int32 key_idx=1, int32 val=2)", rows[0]);
+ }
+ } else {
+ // De-dup, regardless of whether deleted rows are included or not.
+ ASSERT_EQ(1, rows.size());
+ EXPECT_EQ("(int64 key=1, int32 key_idx=1, int32 val=2, is_deleted deleted=false)", rows[0]);
+ }
}
} // namespace tablet
diff --git a/src/kudu/tablet/rowset.cc b/src/kudu/tablet/rowset.cc
index 5bc4e09..beefc8a 100644
--- a/src/kudu/tablet/rowset.cc
+++ b/src/kudu/tablet/rowset.cc
@@ -103,7 +103,7 @@ Status DuplicatingRowSet::NewRowIterator(const RowIteratorOptions& opts,
switch (opts.order) {
case ORDERED:
- *out = NewMergeIterator(std::move(iters));
+ *out = NewMergeIterator(MergeIteratorOptions(opts.include_deleted_rows), std::move(iters));
break;
case UNORDERED:
*out = NewUnionIterator(std::move(iters));
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index 05d4d2f..5a86c8f 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -2468,7 +2468,7 @@ Status Tablet::Iterator::Init(ScanSpec *spec) {
switch (opts_.order) {
case ORDERED:
- iter_ = NewMergeIterator(std::move(iters));
+ iter_ = NewMergeIterator(MergeIteratorOptions(opts_.include_deleted_rows), std::move(iters));
break;
case UNORDERED:
default: