You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/11/30 17:07:41 UTC
[1/5] kudu git commit: Don't output unobservable rows from the
MemRowset
Repository: kudu
Updated Branches:
refs/heads/master 8ff4d1211 -> eb4b46b80
Don't output unobservable rows from the MemRowset
In some rare cases we might have a row that is inserted and deleted
in the same operation, meaning the insert and delete have the same
timestamp. In this rare case we might run into trouble later on with
compactions as there is a sequence of operations that that produce
two ghost rows which are uncomparable in terms of recency.
The sequence is as follows:
Insert Row at 1 -> Row goes in the MRS
Flush -> Row now lives in RS1
Delete Row at 3 -> Row is now a ghost in RS1
Insert Row at 3 -> Row goes in the MRS again
Delete Row at 3 -> Row is deleted from the MRS
Flush -> Row is now a ghost in RS2
Major delta compact RS1 -> GC all before 3
Major delta compact RS2 -> GC all before 3
If RS1 and RS2 now get compacted together there is no way to
distinguish Ghost 1 from Ghost 2 and to build a correct history
for the row.
The point is that we can't trust UNDOs to break ties, in terms of
recency, between two ghosts as they might have been removed by
delta compactions. However we can always trust REDO delete/reinsert
to remain.
This adds a small test to compaction-test that makes sure that
a row that is inserted and deleted in the same transaction doesn't
appear in the compaction input.
FuzzTest and TestRandomAccess would fail for the following patch
(KUDU-237 (part 2)) without this change.
Change-Id: Ieab52a1aa68494218f91f3acd31ef8ddf352bd57
Reviewed-on: http://gerrit.cloudera.org:8080/4994
Tested-by: Kudu Jenkins
Reviewed-by: David Ribeiro Alves <dr...@apache.org>
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/ecff49af
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/ecff49af
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/ecff49af
Branch: refs/heads/master
Commit: ecff49afb95de2cea37987797fe0c115920d2a41
Parents: 8ff4d12
Author: David Alves <dr...@apache.org>
Authored: Mon Nov 7 23:00:59 2016 -0800
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Wed Nov 30 10:14:19 2016 +0000
----------------------------------------------------------------------
src/kudu/tablet/compaction-test.cc | 208 +++++++++++++++++++++++---------
src/kudu/tablet/compaction.cc | 39 +++++-
2 files changed, 185 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/ecff49af/src/kudu/tablet/compaction-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction-test.cc b/src/kudu/tablet/compaction-test.cc
index 5b8d160..3d23c29 100644
--- a/src/kudu/tablet/compaction-test.cc
+++ b/src/kudu/tablet/compaction-test.cc
@@ -97,6 +97,11 @@ class TestCompaction : public KuduRowSetTest {
void InsertRow(MemRowSet *mrs, int row_key, int32_t val) {
ScopedTransaction tx(&mvcc_, clock_->Now());
tx.StartApplying();
+ InsertRowInTransaction(mrs, tx, row_key, val);
+ tx.Commit();
+ }
+
+ void BuildRow(int row_key, int32_t val) {
row_builder_.Reset();
snprintf(key_buf_, sizeof(key_buf_), kRowKeyFormat, row_key);
row_builder_.AddString(Slice(key_buf_));
@@ -106,6 +111,13 @@ class TestCompaction : public KuduRowSetTest {
} else {
row_builder_.AddNull();
}
+ }
+
+ void InsertRowInTransaction(MemRowSet *mrs,
+ const ScopedTransaction& txn,
+ int row_key,
+ int32_t val) {
+ BuildRow(row_key, val);
if (!mrs->schema().Equals(row_builder_.schema())) {
// The MemRowSet is not projecting the row, so must be done by the caller
RowProjector projector(&row_builder_.schema(), &mrs->schema());
@@ -113,12 +125,11 @@ class TestCompaction : public KuduRowSetTest {
ContiguousRow dst_row(&mrs->schema(), rowbuf);
ASSERT_OK_FAST(projector.Init());
ASSERT_OK_FAST(projector.ProjectRowForWrite(row_builder_.row(),
- &dst_row, static_cast<Arena*>(nullptr)));
- ASSERT_OK_FAST(mrs->Insert(tx.timestamp(), ConstContiguousRow(dst_row), op_id_));
+ &dst_row, static_cast<Arena*>(nullptr)));
+ ASSERT_OK_FAST(mrs->Insert(txn.timestamp(), ConstContiguousRow(dst_row), op_id_));
} else {
- ASSERT_OK_FAST(mrs->Insert(tx.timestamp(), row_builder_.row(), op_id_));
+ ASSERT_OK_FAST(mrs->Insert(txn.timestamp(), row_builder_.row(), op_id_));
}
- tx.Commit();
}
// Update n_rows rows of data.
@@ -128,70 +139,90 @@ class TestCompaction : public KuduRowSetTest {
// Note that this is the opposite of InsertRow() above, so that the updates
// flop NULL to non-NULL and vice versa.
void UpdateRows(RowSet *rowset, int n_rows, int delta, int32_t new_val) {
- char keybuf[256];
- faststring update_buf;
- ColumnId col_id = schema_.column_id(schema_.find_column("val"));
- ColumnId nullable_col_id = schema_.column_id(schema_.find_column("nullable_val"));
for (uint32_t i = 0; i < n_rows; i++) {
SCOPED_TRACE(i);
- ScopedTransaction tx(&mvcc_, clock_->Now());
- tx.StartApplying();
- snprintf(keybuf, sizeof(keybuf), kRowKeyFormat, i * 10 + delta);
-
- update_buf.clear();
- RowChangeListEncoder update(&update_buf);
- update.AddColumnUpdate(schema_.column_by_id(col_id), col_id, &new_val);
- if (new_val % 2 == 0) {
- update.AddColumnUpdate(schema_.column_by_id(nullable_col_id),
- nullable_col_id, nullptr);
- } else {
- update.AddColumnUpdate(schema_.column_by_id(nullable_col_id),
- nullable_col_id, &new_val);
- }
-
- RowBuilder rb(schema_.CreateKeyProjection());
- rb.AddString(Slice(keybuf));
- RowSetKeyProbe probe(rb.row());
- ProbeStats stats;
- OperationResultPB result;
- ASSERT_OK(rowset->MutateRow(tx.timestamp(),
- probe,
- RowChangeList(update_buf),
- op_id_,
- &stats,
- &result));
- tx.Commit();
+ UpdateRow(rowset, i * 10 + delta, new_val);
}
}
- void DeleteRows(RowSet *rowset, int n_rows, int delta) {
+ void UpdateRow(RowSet *rowset, int row_key, int32_t new_val) {
+ ScopedTransaction tx(&mvcc_, clock_->Now());
+ tx.StartApplying();
+ UpdateRowInTransaction(rowset, tx, row_key, new_val);
+ tx.Commit();
+ }
+
+ void UpdateRowInTransaction(RowSet *rowset,
+ const ScopedTransaction& txn,
+ int row_key,
+ int32_t new_val) {
+ ColumnId col_id = schema_.column_id(schema_.find_column("val"));
+ ColumnId nullable_col_id = schema_.column_id(schema_.find_column("nullable_val"));
+
char keybuf[256];
faststring update_buf;
+ snprintf(keybuf, sizeof(keybuf), kRowKeyFormat, row_key);
+
+ update_buf.clear();
+ RowChangeListEncoder update(&update_buf);
+ update.AddColumnUpdate(schema_.column_by_id(col_id), col_id, &new_val);
+ if (new_val % 2 == 0) {
+ update.AddColumnUpdate(schema_.column_by_id(nullable_col_id),
+ nullable_col_id, nullptr);
+ } else {
+ update.AddColumnUpdate(schema_.column_by_id(nullable_col_id),
+ nullable_col_id, &new_val);
+ }
+
+ RowBuilder rb(schema_.CreateKeyProjection());
+ rb.AddString(Slice(keybuf));
+ RowSetKeyProbe probe(rb.row());
+ ProbeStats stats;
+ OperationResultPB result;
+ ASSERT_OK(rowset->MutateRow(txn.timestamp(),
+ probe,
+ RowChangeList(update_buf),
+ op_id_,
+ &stats,
+ &result));
+ }
+
+ void DeleteRows(RowSet* rowset, int n_rows) {
+ faststring update_buf;
for (uint32_t i = 0; i < n_rows; i++) {
SCOPED_TRACE(i);
- ScopedTransaction tx(&mvcc_, clock_->Now());
- tx.StartApplying();
- snprintf(keybuf, sizeof(keybuf), kRowKeyFormat, i * 10 + delta);
-
- update_buf.clear();
- RowChangeListEncoder update(&update_buf);
- update.SetToDelete();
-
- RowBuilder rb(schema_.CreateKeyProjection());
- rb.AddString(Slice(keybuf));
- RowSetKeyProbe probe(rb.row());
- ProbeStats stats;
- OperationResultPB result;
- ASSERT_OK(rowset->MutateRow(tx.timestamp(),
- probe,
- RowChangeList(update_buf),
- op_id_,
- &stats,
- &result));
- tx.Commit();
+ DeleteRow(rowset, i * 10);
}
}
+ void DeleteRow(RowSet* rowset, int row_key) {
+ ScopedTransaction tx(&mvcc_, clock_->Now());
+ tx.StartApplying();
+ DeleteRowInTransaction(rowset, tx, row_key);
+ tx.Commit();
+ }
+
+ void DeleteRowInTransaction(RowSet *rowset, const ScopedTransaction& txn, int row_key) {
+ char keybuf[256];
+ faststring update_buf;
+ snprintf(keybuf, sizeof(keybuf), kRowKeyFormat, row_key);
+ update_buf.clear();
+ RowChangeListEncoder update(&update_buf);
+ update.SetToDelete();
+
+ RowBuilder rb(schema_.CreateKeyProjection());
+ rb.AddString(Slice(keybuf));
+ RowSetKeyProbe probe(rb.row());
+ ProbeStats stats;
+ OperationResultPB result;
+ ASSERT_OK(rowset->MutateRow(txn.timestamp(),
+ probe,
+ RowChangeList(update_buf),
+ op_id_,
+ &stats,
+ &result));
+ }
+
// Iterate over the given compaction input, stringifying and dumping each
// yielded row to *out
void IterateInput(CompactionInput *input, vector<string> *out) {
@@ -512,7 +543,7 @@ TEST_F(TestCompaction, TestDuplicatedGhostRowsDontSurviveCompaction) {
}
// Now delete the rows, this will make the rs report them as deleted and
// so we would reinsert them into the MRS.
- DeleteRows(rs1.get(), 10, 0);
+ DeleteRows(rs1.get(), 10);
shared_ptr<DiskRowSet> rs2;
{
@@ -524,7 +555,7 @@ TEST_F(TestCompaction, TestDuplicatedGhostRowsDontSurviveCompaction) {
FlushMRSAndReopenNoRoll(*mrs, schema_, &rs2);
ASSERT_NO_FATAL_FAILURE();
}
- DeleteRows(rs2.get(), 10, 0);
+ DeleteRows(rs2.get(), 10);
shared_ptr<DiskRowSet> rs3;
{
@@ -566,6 +597,67 @@ TEST_F(TestCompaction, TestDuplicatedGhostRowsDontSurviveCompaction) {
"@60(DELETE)]; Redo Mutations: [];", out[9]);
}
+// Test case that inserts and deletes a row in the same transaction and makes sure
+// the row isn't on the compaction input.
+TEST_F(TestCompaction, TestMRSCompactionDoesntOutputUnobservableRows) {
+ // Insert a row in the mrs and flush it.
+ shared_ptr<DiskRowSet> rs1;
+ {
+ shared_ptr<MemRowSet> mrs;
+ ASSERT_OK(MemRowSet::Create(0, schema_, log_anchor_registry_.get(),
+ mem_trackers_.tablet_tracker, &mrs));
+ InsertRow(mrs.get(), 1, 1);
+ FlushMRSAndReopenNoRoll(*mrs, schema_, &rs1);
+ ASSERT_NO_FATAL_FAILURE();
+ }
+
+ // Now make the row a ghost in rs1 in the same transaction as we reinsert it in the mrs then
+ // flush it. Also insert another row so that the row set isn't completely empty (otherwise
+ // it would disappear on flush).
+ shared_ptr<DiskRowSet> rs2;
+ {
+ shared_ptr<MemRowSet> mrs;
+ ASSERT_OK(MemRowSet::Create(1, schema_, log_anchor_registry_.get(),
+ mem_trackers_.tablet_tracker, &mrs));
+ ScopedTransaction tx(&mvcc_, clock_->Now());
+ tx.StartApplying();
+ DeleteRowInTransaction(rs1.get(), tx, 1);
+ InsertRowInTransaction(mrs.get(), tx, 1, 2);
+ UpdateRowInTransaction(mrs.get(), tx, 1, 3);
+ DeleteRowInTransaction(mrs.get(), tx, 1);
+
+ InsertRowInTransaction(mrs.get(), tx, 2, 0);
+ tx.Commit();
+ FlushMRSAndReopenNoRoll(*mrs, schema_, &rs2);
+ ASSERT_NO_FATAL_FAILURE();
+ }
+
+ MvccSnapshot all_snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions();
+
+ gscoped_ptr<CompactionInput> rs1_input;
+ ASSERT_OK(CompactionInput::Create(*rs1, &schema_, all_snap, &rs1_input));
+
+ gscoped_ptr<CompactionInput> rs2_input;
+ ASSERT_OK(CompactionInput::Create(*rs2, &schema_, all_snap, &rs2_input));
+
+ vector<shared_ptr<CompactionInput>> to_merge;
+ to_merge.push_back(shared_ptr<CompactionInput>(rs1_input.release()));
+ to_merge.push_back(shared_ptr<CompactionInput>(rs2_input.release()));
+
+ gscoped_ptr<CompactionInput> merged(CompactionInput::Merge(to_merge, &schema_));
+
+ // Make sure the unobservable version of the row that was inserted and deleted in the MRS
+ // in the same transaction doesn't show up in the compaction input.
+ vector<string> out;
+ IterateInput(merged.get(), &out);
+ EXPECT_EQ(out.size(), 2);
+ EXPECT_EQ("RowIdxInBlock: 0; Base: (string key=hello 00000001, int32 val=1, "
+ "int32 nullable_val=NULL); Undo Mutations: [@1(DELETE)]; Redo Mutations: "
+ "[@2(DELETE)];", out[0]);
+ EXPECT_EQ("RowIdxInBlock: 0; Base: (string key=hello 00000002, int32 val=0, "
+ "int32 nullable_val=0); Undo Mutations: [@2(DELETE)]; Redo Mutations: [];", out[1]);
+}
+
// Test case which doesn't do any merging -- just compacts
// a single input rowset (which may be the memrowset) into a single
// output rowset (on disk).
http://git-wip-us.apache.org/repos/asf/kudu/blob/ecff49af/src/kudu/tablet/compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc
index 826d5b7..bec6615 100644
--- a/src/kudu/tablet/compaction.cc
+++ b/src/kudu/tablet/compaction.cc
@@ -50,6 +50,15 @@ namespace tablet {
namespace {
+// Advances to the last mutation in a mutation list.
+void AdvanceToLastInList(const Mutation** m) {
+ if (*m == nullptr) return;
+ const Mutation* next;
+ while ((next = (*m)->acquire_next()) != nullptr) {
+ *m = next;
+ }
+}
+
// CompactionInput yielding rows and mutations from a MemRowSet.
class MemRowSetCompactionInput : public CompactionInput {
public:
@@ -83,10 +92,11 @@ class MemRowSetCompactionInput : public CompactionInput {
arena_.Reset();
RowChangeListEncoder undo_encoder(&buffer_);
- for (int i = 0; i < num_in_block; i++) {
- // TODO: A copy is performed to make all CompactionInputRow have the same schema
- CompactionInputRow &input_row = block->at(i);
- input_row.row.Reset(row_block_.get(), i);
+ int next_row_index = 0;
+ for (int i = 0; i < num_in_block; ++i) {
+ // TODO(todd): A copy is performed to make all CompactionInputRow have the same schema
+ CompactionInputRow& input_row = block->at(next_row_index);
+ input_row.row.Reset(row_block_.get(), next_row_index);
Timestamp insertion_timestamp;
RETURN_NOT_OK(iter_->GetCurrentRow(&input_row.row,
static_cast<Arena*>(nullptr),
@@ -94,15 +104,36 @@ class MemRowSetCompactionInput : public CompactionInput {
&arena_,
&insertion_timestamp));
+ // Handle the rare case where a row was inserted and deleted in the same operation.
+ // This row can never be observed and should not be compacted/flushed. This saves
+ // us some trouble later on on compactions.
+ // See: MergeCompactionInput::CompareAndMergeDuplicatedRows().
+ if (PREDICT_FALSE(input_row.redo_head != nullptr &&
+ input_row.redo_head->timestamp() == insertion_timestamp)) {
+ // Get the latest mutation.
+ const Mutation* latest = input_row.redo_head;
+ AdvanceToLastInList(&latest);
+ if (latest->changelist().is_delete() &&
+ latest->timestamp() == insertion_timestamp) {
+ iter_->Next();
+ continue;
+ }
+ }
+
// Materialize MRSRow undo insert (delete)
undo_encoder.SetToDelete();
input_row.undo_head = Mutation::CreateInArena(&arena_,
insertion_timestamp,
undo_encoder.as_changelist());
undo_encoder.Reset();
+ ++next_row_index;
iter_->Next();
}
+ if (PREDICT_FALSE(next_row_index < num_in_block)) {
+ block->resize(next_row_index);
+ }
+
has_more_blocks_ = iter_->HasNext();
return Status::OK();
}
[4/5] kudu git commit: Add Reinserts to tablet_history_gc-itest
Posted by to...@apache.org.
Add Reinserts to tablet_history_gc-itest
This adds Reinserts as a new operation to tablet_history_gc-itest.
Reinserts happen on previously deleted rows and, like deletes,
have a change to happen mid-compaction.
This test used to fail frequently before the reinsert patches
are in and now passes.
Ran 500 loops of tablet_history_gc-itest in dist-test in ASAN with:
- KUDU_ALLOW_SLOW_TESTS=1
- --stress_cpu_threads=4
All tests passed. Result:
http://dist-test.cloudera.org//job?job_id=david.alves.1480101254.2208
Change-Id: I6b774b14d79bd7c0c0bf62b585bd70d6d975feaa
Reviewed-on: http://gerrit.cloudera.org:8080/4997
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
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/6f423302
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/6f423302
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/6f423302
Branch: refs/heads/master
Commit: 6f423302d18460355f6b6111ba283a54ca2d9faa
Parents: 10f2ce2
Author: David Alves <dr...@apache.org>
Authored: Mon Nov 7 21:15:45 2016 -0800
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Wed Nov 30 10:14:48 2016 +0000
----------------------------------------------------------------------
.../tablet_history_gc-itest.cc | 158 ++++++++++++++-----
1 file changed, 121 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/6f423302/src/kudu/integration-tests/tablet_history_gc-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/tablet_history_gc-itest.cc b/src/kudu/integration-tests/tablet_history_gc-itest.cc
index e9503c6..0bce2c3 100644
--- a/src/kudu/integration-tests/tablet_history_gc-itest.cc
+++ b/src/kudu/integration-tests/tablet_history_gc-itest.cc
@@ -17,6 +17,7 @@
#include <map>
#include <memory>
+#include <unordered_set>
#include <utility>
#include "kudu/client/client-test-util.h"
@@ -118,6 +119,7 @@ class RandomizedTabletHistoryGcITest : public TabletHistoryGcITest {
kInsert,
kUpdate,
kDelete,
+ kReinsert,
kFlush,
kMergeCompaction,
kRedoDeltaCompaction,
@@ -277,11 +279,11 @@ class RandomizedTabletHistoryGcITest : public TabletHistoryGcITest {
}
MaterializedTestSnapshots snapshots_;
+ std::unordered_set<int32_t> deleted_rows_;
ScannerMap scanners_;
HybridClock* clock_ = nullptr;
Timestamp latest_snapshot_ts_;
- int32_t rows_inserted_ = 0;
int cur_round_ = 0;
};
@@ -308,6 +310,13 @@ class ReupdateHooks : public Tablet::FlushCompactCommonHooks {
CHECK_OK(row.SetInt32(0, row_key));
CHECK_OK(writer.Delete(row));
}
+ for (const MaterializedTestRow& reinsert : reinserts_) {
+ KuduPartialRow row(&client_schema_);
+ CHECK_OK(row.SetInt32(0, reinsert.key));
+ CHECK_OK(row.SetInt32(1, reinsert.int_val));
+ CHECK_OK(row.SetStringCopy(2, reinsert.string_val));
+ CHECK_OK(writer.Insert(row));
+ }
return Status::OK();
}
@@ -319,26 +328,29 @@ class ReupdateHooks : public Tablet::FlushCompactCommonHooks {
deletes_ = std::move(deletes);
}
+ void set_reinserts(vector<MaterializedTestRow> reinserts) {
+ reinserts_ = std::move(reinserts);
+ }
+
void Reset() {
updates_.clear();
deletes_.clear();
+ reinserts_.clear();
}
private:
- enum Action {
- kUpdate,
- kDelete
- };
-
Tablet* const tablet_;
const Schema client_schema_;
vector<MaterializedTestRow> updates_;
vector<int32_t> deletes_;
+ vector<MaterializedTestRow> reinserts_;
};
// Randomized test that attempts to test many arbitrary history GC use cases.
TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
+ const int kSessionTimeoutMillis = 20000;
+
OverrideFlagForSlowTests("test_num_rounds",
Substitute("$0", FLAGS_test_num_rounds * 5));
@@ -376,6 +388,8 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
// Save an empty snapshot at the "beginning of time";
NO_FATALS(SaveSnapshot(MaterializedTestTable(), Timestamp(0)));
+ int32_t rows_inserted = 0;
+
for (cur_round_ = 0; cur_round_ < FLAGS_test_num_rounds; cur_round_++) {
VLOG(1) << "Starting round " << cur_round_;
NO_FATALS(VerifyScannersForRound(cur_round_));
@@ -383,19 +397,17 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
int action = random.Uniform(kNumActions);
switch (action) {
case kInsert: {
- // TODO: Allow for reinsert onto deleted rows after KUDU-237 has been
- // implemented.
int32_t num_rows_to_insert = random.Uniform(1000);
VLOG(1) << "Inserting " << num_rows_to_insert << " rows";
if (num_rows_to_insert == 0) continue;
MaterializedTestTable snapshot = CloneLatestSnapshot();
client::sp::shared_ptr<client::KuduSession> session = client_->NewSession();
- session->SetTimeoutMillis(20000);
+ session->SetTimeoutMillis(kSessionTimeoutMillis);
ASSERT_OK_FAST(session->SetFlushMode(client::KuduSession::MANUAL_FLUSH));
for (int32_t i = 0; i < num_rows_to_insert; i++) {
- int32_t row_key = rows_inserted_;
+ int32_t row_key = rows_inserted;
MaterializedTestRow test_row = { row_key,
static_cast<int32_t>(random.Next()),
Substitute("$0", random.Next()),
@@ -404,20 +416,20 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
KuduPartialRow* row = insert->mutable_row();
ASSERT_OK_FAST(row->SetInt32(0, test_row.key));
ASSERT_OK_FAST(row->SetInt32(1, test_row.int_val));
- ASSERT_OK_FAST(row->SetStringCopy(2, test_row.string_val));
+ ASSERT_OK_FAST(row->SetString(2, test_row.string_val));
ASSERT_OK_FAST(session->Apply(insert.release()));
VLOG(2) << "Inserting row " << StringifyTestRow(test_row);
snapshot[row_key] = std::move(test_row);
- rows_inserted_++;
+ rows_inserted++;
}
FlushSessionOrDie(session);
SaveSnapshot(std::move(snapshot), clock_->Now());
break;
}
case kUpdate: {
- if (rows_inserted_ == 0) continue;
- int32_t num_rows_to_update = random.Uniform(std::min(rows_inserted_, 1000));
+ if (rows_inserted == 0) continue;
+ int32_t num_rows_to_update = random.Uniform(std::min(rows_inserted, 1000));
VLOG(1) << "Updating up to " << num_rows_to_update << " rows";
if (num_rows_to_update == 0) continue;
@@ -431,18 +443,17 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
vector<MaterializedTestRow> updates;
for (int i = 0; i < num_rows_to_update; i++) {
- int32_t row_key = random.Uniform(rows_inserted_);
- MaterializedTestRow test_row = snapshot[row_key];
- CHECK_EQ(row_key, test_row.key) << "Rows inserted: " << rows_inserted_
- << ", row: " << StringifyTestRow(test_row);
- if (test_row.is_deleted == DELETED) continue;
+ int32_t row_key = random.Uniform(rows_inserted);
+ MaterializedTestRow* test_row = &snapshot[row_key];
+ ASSERT_EQ(row_key, test_row->key) << "Rows inserted: " << rows_inserted
+ << ", row: " << StringifyTestRow(*test_row);
+ if (test_row->is_deleted == DELETED) continue;
- test_row.int_val = random.Next();
- test_row.string_val = Substitute("$0", random.Next());
+ test_row->int_val = random.Next();
+ test_row->string_val = Substitute("$0", random.Next());
- VLOG(2) << "Updating row to " << StringifyTestRow(test_row);
- updates.push_back(test_row);
- snapshot[row_key] = std::move(test_row);
+ VLOG(2) << "Updating row to " << StringifyTestRow(*test_row);
+ updates.push_back(*test_row);
}
int rows_updated = updates.size();
@@ -457,7 +468,7 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
tablet->SetFlushCompactCommonHooksForTests(nullptr); // Clear the hook.
} else {
client::sp::shared_ptr<client::KuduSession> session = client_->NewSession();
- session->SetTimeoutMillis(20000);
+ session->SetTimeoutMillis(kSessionTimeoutMillis);
ASSERT_OK_FAST(session->SetFlushMode(client::KuduSession::MANUAL_FLUSH));
for (const MaterializedTestRow& test_row : updates) {
@@ -465,7 +476,7 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
KuduPartialRow* row = update->mutable_row();
ASSERT_OK_FAST(row->SetInt32(0, test_row.key));
ASSERT_OK_FAST(row->SetInt32(1, test_row.int_val));
- ASSERT_OK_FAST(row->SetStringCopy(2, test_row.string_val));
+ ASSERT_OK_FAST(row->SetString(2, test_row.string_val));
ASSERT_OK_FAST(session->Apply(update.release()));
}
FlushSessionOrDie(session);
@@ -475,8 +486,8 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
break;
}
case kDelete: {
- if (rows_inserted_ == 0) continue;
- int32_t num_rows_to_delete = random.Uniform(std::min(rows_inserted_, 1000));
+ if (rows_inserted == 0) continue;
+ int32_t num_rows_to_delete = random.Uniform(std::min(rows_inserted, 1000));
VLOG(1) << "Deleting up to " << num_rows_to_delete << " rows";
if (num_rows_to_delete == 0) continue;
@@ -490,23 +501,25 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
vector<int32_t> deletes;
for (int i = 0; i < num_rows_to_delete; i++) {
- int32_t row_key = random.Uniform(rows_inserted_);
- MaterializedTestRow test_row = snapshot[row_key];
- CHECK_EQ(row_key, test_row.key);
+ int32_t row_key = random.Uniform(rows_inserted);
+ MaterializedTestRow* test_row = &snapshot[row_key];
+ CHECK_EQ(row_key, test_row->key);
- if (test_row.is_deleted == DELETED) {
- VLOG(2) << "Row " << test_row.key << " is already deleted";
+ if (test_row->is_deleted == DELETED) {
+ VLOG(2) << "Row " << test_row->key << " is already deleted";
continue;
}
- VLOG(2) << "Deleting row " << row_key;
+
+ test_row->is_deleted = DELETED;
+ VLOG(2) << "Deleting row " << StringifyTestRow(*test_row);
deletes.push_back(row_key);
- test_row.is_deleted = DELETED;
- snapshot[row_key] = std::move(test_row);
}
int rows_deleted = deletes.size();
if (rows_deleted == 0) continue;
+ deleted_rows_.insert(deletes.begin(), deletes.end());
+
if (force_reupdate_missed_deltas) {
std::shared_ptr<ReupdateHooks> hooks =
std::make_shared<ReupdateHooks>(tablet, GetSimpleTestSchema());
@@ -516,7 +529,7 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
tablet->SetFlushCompactCommonHooksForTests(nullptr); // Clear the hook.
} else {
client::sp::shared_ptr<client::KuduSession> session = client_->NewSession();
- session->SetTimeoutMillis(20000);
+ session->SetTimeoutMillis(kSessionTimeoutMillis);
ASSERT_OK_FAST(session->SetFlushMode(client::KuduSession::MANUAL_FLUSH));
for (int32_t row_key : deletes) {
@@ -531,6 +544,77 @@ TEST_F(RandomizedTabletHistoryGcITest, TestRandomHistoryGCWorkload) {
VLOG(1) << "Deleted " << rows_deleted << " rows";
break;
}
+ case kReinsert: {
+ if (rows_inserted == 0) continue;
+ int32_t max_rows_to_reinsert = random.Uniform(std::min(rows_inserted, 1000));
+ VLOG(1) << "Reinserting up to " << max_rows_to_reinsert << " rows";
+ if (max_rows_to_reinsert == 0) continue;
+ int num_deleted_rows = deleted_rows_.size();
+ if (num_deleted_rows == 0) continue;
+
+
+ // 5% chance to reupdate while also forcing a full compaction.
+ bool force_reupdate_missed_deltas = random.OneIn(20);
+ if (force_reupdate_missed_deltas) {
+ VLOG(1) << "Forcing a reupdate of missed deltas";
+ }
+
+ MaterializedTestTable snapshot = CloneLatestSnapshot();
+
+ vector<MaterializedTestRow> reinserts;
+ for (int i = 0; i < max_rows_to_reinsert; i++) {
+ const int32_t row_key = *std::next(deleted_rows_.begin(),
+ random.Uniform(num_deleted_rows));
+ MaterializedTestRow* test_row = &snapshot[row_key];
+ ASSERT_EQ(row_key, test_row->key);
+
+ if (test_row->is_deleted == NOT_DELETED) {
+ VLOG(2) << "Row " << test_row->key << " is already reinserted";
+ continue;
+ }
+
+ test_row->int_val = random.Next();
+ test_row->string_val = Substitute("$0", random.Next());
+ test_row->is_deleted = NOT_DELETED;
+
+ VLOG(2) << "Reinserting row " << StringifyTestRow(*test_row);
+ reinserts.push_back(*test_row);
+ }
+
+ if (reinserts.empty()) continue;
+
+ for (const MaterializedTestRow& test_row : reinserts) {
+ deleted_rows_.erase(test_row.key);
+ }
+
+ int num_reinserted = reinserts.size();
+
+ if (force_reupdate_missed_deltas) {
+ std::shared_ptr<ReupdateHooks> hooks =
+ std::make_shared<ReupdateHooks>(tablet, GetSimpleTestSchema());
+ hooks->set_reinserts(std::move(reinserts));
+ tablet->SetFlushCompactCommonHooksForTests(hooks);
+ ASSERT_OK(tablet->Compact(Tablet::FORCE_COMPACT_ALL));
+ tablet->SetFlushCompactCommonHooksForTests(nullptr); // Clear the hook.
+ } else {
+ client::sp::shared_ptr<client::KuduSession> session = client_->NewSession();
+ session->SetTimeoutMillis(kSessionTimeoutMillis);
+ ASSERT_OK_FAST(session->SetFlushMode(client::KuduSession::MANUAL_FLUSH));
+
+ for (const MaterializedTestRow& test_row : reinserts) {
+ unique_ptr<client::KuduInsert> reinsert(table->NewInsert());
+ KuduPartialRow* row = reinsert->mutable_row();
+ ASSERT_OK_FAST(row->SetInt32(0, test_row.key));
+ ASSERT_OK_FAST(row->SetInt32(1, test_row.int_val));
+ ASSERT_OK_FAST(row->SetString(2, test_row.string_val));
+ ASSERT_OK_FAST(session->Apply(reinsert.release()));
+ }
+ FlushSessionOrDie(session);
+ }
+ SaveSnapshot(std::move(snapshot), clock_->Now());
+ VLOG(1) << "Reinserted " << num_reinserted << " rows";
+ break;
+ }
case kFlush: {
if (random.OneIn(2)) {
VLOG(1) << "Flushing tablet";
[2/5] kudu git commit: KUDU-237 (part 2) - Add support for REINSERT
in delta files
Posted by to...@apache.org.
KUDU-237 (part 2) - Add support for REINSERT in delta files
This patch goes the final mile in adding support for REINSERTs in
delta files. It does various things to achieve this:
- Transforms REDO DELETE/REINSERT mutation pairs into UNDO
REINSERT/DELETE mutation pairs, leaving at the most a REDO
delete.
- Merges ghost rows on compaction: When duplicated rows are
found a new algorithm finds out which one is the most recent
and adds the other one as a 'previous_ghost'. This can happen
for an arbitrary number of ghost rows. Noteworthy is that setting
previous versions requires a copy. The two rows are in different
RowBlocks (for row data) and Arenas (for mutations) and it is
not guaranteed that the previous ghost will suvive by the time the
row that points to it is processed.
- Adds new test to tablet-test and changes a test in
compaction-test proving that this works.
- Adds a new test to compaction-test that creates several layers
of overlapping rowsets where each layer has one rowset less than
the previous one. This creates a mix of duplicated (up to 10
different ghosts) and unique rows. The test then compacts the
rowsets a few at a time and makes sure the histories are correct.
Follow up patches will add new itests that test this functionality
even more broadly.
Ran 500 loops of mt-tablet-test and compaction-test
in dist-test with "KUDU_ALLOW_SLOW_TESTS=1" and
"--stress_cpu_threads=4". All tests passed. Results:
compaction-test: http://dist-test.cloudera.org//job?job_id=david.alves.1480028902.31041
mt-tablet-test: http://dist-test.cloudera.org//job?job_id=david.alves.1480028908.31125
Change-Id: Ie1173b2bea721b376f2b6049be20f57307582c47
Reviewed-on: http://gerrit.cloudera.org:8080/4995
Tested-by: Kudu Jenkins
Reviewed-by: David Ribeiro Alves <dr...@apache.org>
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/abab2ced
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/abab2ced
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/abab2ced
Branch: refs/heads/master
Commit: abab2ced14c2166ca057573b78b738b0ebc113b2
Parents: ecff49a
Author: David Alves <dr...@apache.org>
Authored: Mon Nov 7 22:39:36 2016 -0800
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Wed Nov 30 10:14:29 2016 +0000
----------------------------------------------------------------------
src/kudu/common/row_changelist.h | 18 +-
src/kudu/tablet/compaction-test.cc | 199 +++++++++++++++-
src/kudu/tablet/compaction.cc | 410 +++++++++++++++++++++++++-------
src/kudu/tablet/compaction.h | 12 +-
src/kudu/tablet/delta_store.cc | 1 -
src/kudu/tablet/deltafile.cc | 53 ++---
src/kudu/tablet/deltafile.h | 6 +-
src/kudu/tablet/tablet-test.cc | 63 +++++
8 files changed, 631 insertions(+), 131 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/common/row_changelist.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_changelist.h b/src/kudu/common/row_changelist.h
index ed5f536..5c1987d 100644
--- a/src/kudu/common/row_changelist.h
+++ b/src/kudu/common/row_changelist.h
@@ -199,6 +199,13 @@ class RowChangeListEncoder {
(type_ == RowChangeList::kReinsert || type_ == RowChangeList::kUpdate);
}
+ // Internal version of AddColumnUpdate which does not set type, allowing
+ // it to work for both REINSERT and UPDATE.
+ // Exposed for tests.
+ void EncodeColumnMutation(const ColumnSchema& col_schema,
+ int col_id,
+ const void* cell_ptr);
+
private:
FRIEND_TEST(TestRowChangeList, TestInvalid_SetNullForNonNullableColumn);
FRIEND_TEST(TestRowChangeList, TestInvalid_SetWrongSizeForIntColumn);
@@ -217,12 +224,6 @@ class RowChangeListEncoder {
DCHECK_EQ(type_, type);
}
- // Internal version of AddColumnUpdate which does not set type, allowing
- // it to work for both REINSERT and UPDATE.
- void EncodeColumnMutation(const ColumnSchema& col_schema,
- int col_id,
- const void* cell_ptr);
-
// Add a column mutation by a raw value. This allows copying RCLs
// from one file to another without having any awareness of schema.
//
@@ -306,12 +307,11 @@ class RowChangeListDecoder {
return type_;
}
- // Append an entry to *column_ids for each column that is updated
- // in this RCL.
+ // Append an entry to *column_ids for each column that is mutated in this RCL.
// This 'consumes' the remainder of the encoded RowChangeList.
Status GetIncludedColumnIds(std::vector<ColumnId>* column_ids) {
+ DCHECK(is_update() || is_reinsert());
column_ids->clear();
- DCHECK(is_update());
while (HasNext()) {
DecodedUpdate dec;
RETURN_NOT_OK(DecodeNext(&dec));
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/compaction-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction-test.cc b/src/kudu/tablet/compaction-test.cc
index 3d23c29..c90caa2 100644
--- a/src/kudu/tablet/compaction-test.cc
+++ b/src/kudu/tablet/compaction-test.cc
@@ -70,6 +70,7 @@ class TestCompaction : public KuduRowSetTest {
: KuduRowSetTest(CreateSchema()),
op_id_(consensus::MaximumOpId()),
row_builder_(schema_),
+ arena_(32*1024, 128*1024),
clock_(server::LogicalClock::CreateStartingAt(Timestamp::kInitialTimestamp)),
mvcc_(clock_),
log_anchor_registry_(new log::LogAnchorRegistry()) {
@@ -423,12 +424,19 @@ class TestCompaction : public KuduRowSetTest {
}
}
+ // Helpers for building an expected row history.
+ void AddExpectedDelete(Mutation** current_head, Timestamp ts = Timestamp::kInvalidTimestamp);
+ void AddExpectedUpdate(Mutation** current_head, int32_t val);
+ void AddExpectedReinsert(Mutation** current_head, int32_t val);
+ void AddUpdateAndDelete(RowSet* rs, CompactionInputRow* row, int row_id, int32_t val);
+
protected:
OpId op_id_;
RowBuilder row_builder_;
char key_buf_[256];
- scoped_refptr<server::Clock> clock_;
+ Arena arena_;
+ scoped_refptr<server::LogicalClock> clock_;
MvccManager mvcc_;
scoped_refptr<LogAnchorRegistry> log_anchor_registry_;
@@ -530,8 +538,9 @@ TEST_F(TestCompaction, TestRowSetInput) {
}
// Tests that the same rows, duplicated in three DRSs, ghost in two of them
-// appears only once on the compaction output
-TEST_F(TestCompaction, TestDuplicatedGhostRowsDontSurviveCompaction) {
+// appears only once on the compaction output but that the resulting row
+// includes reinserts for the ghost and all its mutations.
+TEST_F(TestCompaction, TestDuplicatedGhostRowsMerging) {
shared_ptr<DiskRowSet> rs1;
{
shared_ptr<MemRowSet> mrs;
@@ -591,10 +600,190 @@ TEST_F(TestCompaction, TestDuplicatedGhostRowsDontSurviveCompaction) {
ASSERT_EQ(out.size(), 10);
EXPECT_EQ("RowIdxInBlock: 0; Base: (string key=hello 00000000, int32 val=2, "
"int32 nullable_val=NULL); Undo Mutations: [@61(SET val=0, nullable_val=0), "
- "@51(DELETE)]; Redo Mutations: [];", out[0]);
+ "@51(DELETE), @41(REINSERT val=1, nullable_val=1), @31(SET val=0, nullable_val=0), "
+ "@21(DELETE), @11(REINSERT val=0, nullable_val=0), @1(DELETE)]; "
+ "Redo Mutations: [];", out[0]);
EXPECT_EQ("RowIdxInBlock: 9; Base: (string key=hello 00000090, int32 val=2, "
"int32 nullable_val=NULL); Undo Mutations: [@70(SET val=9, nullable_val=NULL), "
- "@60(DELETE)]; Redo Mutations: [];", out[9]);
+ "@60(DELETE), @50(REINSERT val=1, nullable_val=1), @40(SET val=9, "
+ "nullable_val=NULL), @30(DELETE), @20(REINSERT val=9, nullable_val=NULL), "
+ "@10(DELETE)]; Redo Mutations: [];", out[9]);
+}
+
+void TestCompaction::AddExpectedDelete(Mutation** current_head, Timestamp ts) {
+ faststring buf;
+ RowChangeListEncoder enc(&buf);
+ enc.SetToDelete();
+ if (ts == Timestamp::kInvalidTimestamp) ts = Timestamp(clock_->GetCurrentTime());
+ Mutation* mutation = Mutation::CreateInArena(&arena_,
+ ts,
+ enc.as_changelist());
+ mutation->set_next(*current_head);
+ *current_head = mutation;
+}
+
+void TestCompaction::AddExpectedUpdate(Mutation** current_head, int32_t val) {
+ faststring buf;
+ RowChangeListEncoder enc(&buf);
+ enc.SetToUpdate();
+ enc.AddColumnUpdate(schema_.column(1), schema_.column_id(1), &val);
+ if (val % 2 == 0) {
+ enc.AddColumnUpdate(schema_.column(2), schema_.column_id(2), &val);
+ } else {
+ enc.AddColumnUpdate(schema_.column(2), schema_.column_id(2), nullptr);
+ }
+ Mutation* mutation = Mutation::CreateInArena(&arena_,
+ Timestamp(clock_->GetCurrentTime()),
+ enc.as_changelist());
+ mutation->set_next(*current_head);
+ *current_head = mutation;
+}
+
+void TestCompaction::AddExpectedReinsert(Mutation** current_head, int32_t val) {
+ faststring buf;
+ RowChangeListEncoder enc(&buf);
+ enc.SetToReinsert();
+ enc.EncodeColumnMutation(schema_.column(1), schema_.column_id(1), &val);
+ if (val % 2 == 1) {
+ enc.EncodeColumnMutation(schema_.column(2), schema_.column_id(2), &val);
+ } else {
+ enc.EncodeColumnMutation(schema_.column(2), schema_.column_id(2), nullptr);
+ }
+ Mutation* mutation = Mutation::CreateInArena(&arena_, Timestamp(clock_->GetCurrentTime()),
+ enc.as_changelist());
+ mutation->set_next(*current_head);
+ *current_head = mutation;
+}
+
+void TestCompaction::AddUpdateAndDelete(RowSet* rs, CompactionInputRow* row, int row_id,
+ int32_t val) {
+ UpdateRow(rs, row_id, val);
+ // Expect an UNDO update for the update.
+ AddExpectedUpdate(&row->undo_head, row_id);
+
+ DeleteRow(rs, row_id);
+ // Expect an UNDO reinsert for the delete.
+ AddExpectedReinsert(&row->undo_head, val);
+}
+
+// Build several layers of overlapping rowsets with many ghost rows.
+// Repeatedly merge all the generated RowSets until we are left with a single RowSet, then make
+// sure that its history matches our expected history.
+//
+// There are 'kBaseNumRowSets' layers of overlapping rowsets, each level has one less rowset and
+// thus less rows. This is meant to exercise a normal-ish path where there are both duplicated and
+// unique rows per merge while at the same time making sure that some of the rows are duplicated
+// many times.
+//
+// The verification is performed against a vector of expected CompactionInputRow that we build
+// as we insert/update/delete.
+TEST_F(TestCompaction, TestDuplicatedRowsRandomCompaction) {
+ const int kBaseNumRowSets = 10;
+ const int kNumRowsPerRowSet = 10;
+
+ int total_num_rows = kBaseNumRowSets * kNumRowsPerRowSet;
+
+ MvccSnapshot all_snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions();
+
+ vector<CompactionInputRow> expected_rows(total_num_rows);
+ vector<shared_ptr<DiskRowSet>> row_sets;
+
+ // Create a vector of ids for rows and fill it for the first layer.
+ vector<int> row_ids(total_num_rows);
+ std::iota(row_ids.begin(), row_ids.end(), 0);
+
+ SeedRandom();
+ int rs_id = 0;
+ for (int i = 0; i < kBaseNumRowSets; ++i) {
+ int num_rowsets_in_layer = kBaseNumRowSets - i;
+ size_t row_idx = 0;
+ for (int j = 0; j < num_rowsets_in_layer; ++j) {
+ shared_ptr<MemRowSet> mrs;
+ ASSERT_OK(MemRowSet::Create(rs_id, schema_, log_anchor_registry_.get(),
+ mem_trackers_.tablet_tracker, &mrs));
+
+ // For even rows, insert, update and delete them in the mrs.
+ for (int k = 0; k < kNumRowsPerRowSet; ++k) {
+ int row_id = row_ids[row_idx + k];
+ CompactionInputRow* row = &expected_rows[row_id];
+ InsertRow(mrs.get(), row_id, row_id);
+ // Expect an UNDO delete for the insert.
+ AddExpectedDelete(&row->undo_head);
+ if (row_id % 2 == 0) AddUpdateAndDelete(mrs.get(), row, row_id, row_id + i + 1);
+ }
+ shared_ptr<DiskRowSet> drs;
+ FlushMRSAndReopenNoRoll(*mrs, schema_, &drs);
+ // For odd rows, update them and delete them in the drs.
+ for (int k = 0; k < kNumRowsPerRowSet; ++k) {
+ int row_id = row_ids[row_idx];
+ CompactionInputRow* row = &expected_rows[row_id];
+ if (row_id % 2 == 1) AddUpdateAndDelete(drs.get(), row, row_id, row_id + i + 1);
+ row_idx++;
+ }
+ row_sets.push_back(drs);
+ rs_id++;
+ }
+ // For the next layer remove one rowset worth of rows at random.
+ for (int j = 0; j < kNumRowsPerRowSet; ++j) {
+ int to_remove = rand() % row_ids.size();
+ row_ids.erase(row_ids.begin() + to_remove);
+ }
+
+ }
+
+ RowBlock block(schema_, kBaseNumRowSets * kNumRowsPerRowSet, &arena_);
+ // Go through the expected compaction input rows, flip the last undo into a redo and
+ // build the base. This will give us the final version that we'll expect the result
+ // of the real compaction to match.
+ for (int i = 0; i < expected_rows.size(); ++i) {
+ CompactionInputRow* row = &expected_rows[i];
+ Mutation* reinsert = row->undo_head;
+ row->undo_head = reinsert->next();
+ row->row = block.row(i);
+ BuildRow(i, i);
+ CopyRow(row_builder_.row(), &row->row, &arena_);
+ RowChangeListDecoder redo_decoder(reinsert->changelist());
+ CHECK_OK(redo_decoder.Init());
+ faststring buf;
+ RowChangeListEncoder dummy(&buf);
+ dummy.SetToUpdate();
+ redo_decoder.MutateRowAndCaptureChanges(&row->row, &arena_, &dummy);
+ AddExpectedDelete(&row->redo_head, reinsert->timestamp());
+ }
+
+ vector<shared_ptr<CompactionInput>> inputs;
+ for (auto& row_set : row_sets) {
+ gscoped_ptr<CompactionInput> ci;
+ CHECK_OK(row_set->NewCompactionInput(&schema_, all_snap, &ci));
+ inputs.push_back(shared_ptr<CompactionInput>(ci.release()));
+ }
+
+ // Compact the row sets by picking a few at random until we're left with just one.
+ while (row_sets.size() > 1) {
+ std::random_shuffle(row_sets.begin(), row_sets.end());
+ // Merge between 2 and 4 row sets.
+ int num_rowsets_to_merge = std::min(rand() % 3 + 2, static_cast<int>(row_sets.size()));
+ vector<shared_ptr<DiskRowSet>> to_merge;
+ for (int i = 0; i < num_rowsets_to_merge; ++i) {
+ to_merge.push_back(row_sets.back());
+ row_sets.pop_back();
+ }
+ shared_ptr<DiskRowSet> result;
+ CompactAndReopenNoRoll(to_merge, schema_, &result);
+ row_sets.push_back(result);
+ }
+
+ vector<string> out;
+ gscoped_ptr<CompactionInput> ci;
+ CHECK_OK(row_sets[0]->NewCompactionInput(&schema_, all_snap, &ci));
+ IterateInput(ci.get(), &out);
+
+ // Finally go through the final compaction input and through the expected one and make sure
+ // they match.
+ ASSERT_EQ(expected_rows.size(), out.size());
+ for (int i = 0; i < expected_rows.size(); ++i) {
+ EXPECT_EQ(CompactionInputRowToString(expected_rows[i]), out[i]);
+ }
}
// Test case that inserts and deletes a row in the same transaction and makes sure
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc
index bec6615..45f5772 100644
--- a/src/kudu/tablet/compaction.cc
+++ b/src/kudu/tablet/compaction.cc
@@ -251,6 +251,87 @@ class DiskRowSetCompactionInput : public CompactionInput {
};
};
+// Compares two duplicate rows before compaction (and before the REDO->UNDO transformation).
+// Returns -1 if 'left' is less recent than 'right', 1 otherwise. Never returns 0.
+int CompareDuplicatedRows(const CompactionInputRow& left,
+ const CompactionInputRow& right) {
+ const Mutation* left_last = left.redo_head;
+ const Mutation* right_last = right.redo_head;
+ AdvanceToLastInList(&left_last);
+ AdvanceToLastInList(&right_last);
+
+ if (left.redo_head == nullptr) {
+ // left must still be alive, meaning right must have at least a DELETE redo.
+ DCHECK(right_last != nullptr);
+ DCHECK(right_last->changelist().is_delete());
+ return 1;
+ }
+ if (right.redo_head == nullptr) {
+ // right must still be alive, meaning left must have at least a DELETE redo.
+ DCHECK(left_last != nullptr);
+ DCHECK(left_last->changelist().is_delete());
+ return -1;
+ }
+
+ // Duplicated rows usually have disjoint redo histories, meaning the first mutation
+ // should be enough for the sake of determining the most recent row in most cases.
+ int ret = left.redo_head->timestamp().CompareTo(right.redo_head->timestamp());
+
+ if (ret > 0) {
+ return ret;
+ }
+
+ if (PREDICT_TRUE(ret < 0)) {
+ return ret;
+ }
+
+ // In case the histories aren't disjoint it must be because one row was deleted in the
+ // the same operation that inserted the other one, which was then mutated.
+ // For instance this is a valid history with non-disjoint REDO histories:
+ // -- Row 'a' lives in DRS1
+ // Update row 'a' @ 10
+ // Delete row 'a' @ 10
+ // Insert row 'a' @ 10
+ // -- Row 'a' lives in the MRS
+ // Update row 'a' @ 10
+ // -- Flush the MRS into DRS2
+ // -- Compact DRS1 and DRS2
+ //
+ // We can't have the case here where both 'left' and 'right' have a DELETE as the last
+ // mutation at the same timestamp. This would be troublesome as we would possibly have
+ // no way to decide which version is the most up-to-date (one or both version's undos
+ // might have been garbage collected). See MemRowSetCompactionInput::PrepareBlock().
+
+ // At least one of the rows must have a DELETE REDO as its last redo.
+ CHECK(left_last->changelist().is_delete() || right_last->changelist().is_delete());
+ // We made sure that rows that are inserted and deleted in the same operation can never
+ // be part of a compaction input so they can't both be deletes.
+ CHECK(!(left_last->changelist().is_delete() && right_last->changelist().is_delete()));
+
+ // If 'left' doesn't have a delete then it's the latest version.
+ if (!left_last->changelist().is_delete() && right_last->changelist().is_delete()) {
+ return 1;
+ }
+
+ // ...otherwise it's 'right'.
+ return -1;
+}
+
+void CopyMutations(Mutation* from, Mutation** to, Arena* arena) {
+ Mutation* previous = nullptr;
+ for (const Mutation* cur = from; cur != nullptr; cur = cur->acquire_next()) {
+ Mutation* copy = Mutation::CreateInArena(arena,
+ cur->timestamp(),
+ cur->changelist());
+ if (previous != nullptr) {
+ previous->set_next(copy);
+ } else {
+ *to = copy;
+ }
+ previous = copy;
+ }
+}
+
class MergeCompactionInput : public CompactionInput {
private:
// State kept for each of the inputs.
@@ -267,8 +348,12 @@ class MergeCompactionInput : public CompactionInput {
return pending_idx >= pending.size();
}
- const CompactionInputRow &next() const {
- return pending[pending_idx];
+ CompactionInputRow* next() {
+ return &pending[pending_idx];
+ }
+
+ const CompactionInputRow* next() const {
+ return &pending[pending_idx];
}
void pop_front() {
@@ -289,7 +374,7 @@ class MergeCompactionInput : public CompactionInput {
DCHECK(!empty());
DCHECK(!other.empty());
- return schema.Compare(pending.back().row, other.next().row) < 0;
+ return schema.Compare(pending.back().row, (*other.next()).row) < 0;
}
shared_ptr<CompactionInput> input;
@@ -302,7 +387,8 @@ class MergeCompactionInput : public CompactionInput {
public:
MergeCompactionInput(const vector<shared_ptr<CompactionInput> > &inputs,
const Schema* schema)
- : schema_(schema) {
+ : schema_(schema),
+ num_dup_rows_(0) {
for (const shared_ptr<CompactionInput> &input : inputs) {
gscoped_ptr<MergeState> state(new MergeState);
state->input = input;
@@ -344,7 +430,7 @@ class MergeCompactionInput : public CompactionInput {
while (true) {
int smallest_idx = -1;
- CompactionInputRow smallest;
+ CompactionInputRow* smallest;
// Iterate over the inputs to find the one with the smallest next row.
// It may seem like an O(n lg k) merge using a heap would be more efficient,
@@ -367,40 +453,38 @@ class MergeCompactionInput : public CompactionInput {
smallest = state->next();
continue;
}
- int row_comp = schema_->Compare(state->next().row, smallest.row);
+ int row_comp = schema_->Compare(state->next()->row, smallest->row);
if (row_comp < 0) {
smallest_idx = i;
smallest = state->next();
continue;
}
- // If we found two duplicated rows, we want the row with the highest
- // live version. If they're equal, that can only be because they're both
- // dead, in which case it doesn't matter.
- // TODO: this is going to change with historical REINSERT handling.
+ // If we found two rows with the same key, we want to make the newer one point to the older
+ // one, which must be a ghost.
if (PREDICT_FALSE(row_comp == 0)) {
- int mutation_comp = CompareLatestLiveVersion(state->next(), smallest);
+ int mutation_comp = CompareDuplicatedRows(*state->next(), *smallest);
+ CHECK_NE(mutation_comp, 0);
if (mutation_comp > 0) {
// If the previous smallest row has a highest version that is lower
- // than this one, discard it.
+ // than this one, clone it as the previous version and discard the original.
+ RETURN_NOT_OK(SetPreviousGhost(state->next(), smallest, true /* clone */,
+ state->input->PreparedBlockArena()));
states_[smallest_idx]->pop_front();
smallest_idx = i;
smallest = state->next();
continue;
- } else {
- // .. otherwise pop the other one.
- //
- // NOTE: If they're equal, then currently that means that both versions are
- // ghosts. Once we handle REINSERTS, we'll have to figure out which one "comes
- // first" and deal with this properly. For now, we can just pick arbitrarily.
- states_[i]->pop_front();
- continue;
}
+ // .. otherwise copy and pop the other one.
+ RETURN_NOT_OK(SetPreviousGhost(smallest, state->next(), true /* clone */,
+ smallest->row.row_block()->arena()));
+ states_[i]->pop_front();
+ continue;
}
}
DCHECK_GE(smallest_idx, 0);
states_[smallest_idx]->pop_front();
- block->push_back(smallest);
+ block->push_back(*smallest);
}
return Status::OK();
@@ -502,54 +586,194 @@ class MergeCompactionInput : public CompactionInput {
}
}
- // Compare the mutations of two duplicated rows.
- // Returns -1 if latest_version(left) < latest_version(right)
- static int CompareLatestLiveVersion(const CompactionInputRow& left,
- const CompactionInputRow& right) {
- if (left.redo_head == nullptr) {
- // left must still be alive
- DCHECK(right.redo_head != nullptr);
- return 1;
+ // (Deep) clones a compaction input row, copying both the row data, all undo/redo mutations and
+ // all previous ghosts to 'arena'.
+ Status CloneCompactionInputRow(const CompactionInputRow* src,
+ CompactionInputRow** dst,
+ Arena* arena) {
+ CompactionInputRow* copy = arena->NewObject<CompactionInputRow>();
+ copy->row = NewRow();
+ // Copy the row to the arena.
+ RETURN_NOT_OK(CopyRow(src->row, ©->row, arena));
+ // ... along with the redos and undos.
+ CopyMutations(src->redo_head, ©->redo_head, arena);
+ CopyMutations(src->undo_head, ©->undo_head, arena);
+ // Copy previous versions recursively.
+ if (src->previous_ghost != nullptr) {
+ CompactionInputRow* child;
+ RETURN_NOT_OK(CloneCompactionInputRow(src->previous_ghost, &child, arena));
+ copy->previous_ghost = child;
}
- if (right.redo_head == nullptr) {
- DCHECK(left.redo_head != nullptr);
- return -1;
+ *dst = copy;
+ return Status::OK();
+ }
+
+ // Sets the previous ghost row for a CompactionInputRow.
+ // 'must_copy' indicates whether there must be a deep copy (using CloneCompactionInputRow()).
+ Status SetPreviousGhost(CompactionInputRow* older,
+ CompactionInputRow* newer,
+ bool must_copy,
+ Arena* arena) {
+ CHECK(arena != nullptr) << "Arena can't be null";
+ // Check if we already had a previous version and, if yes, whether 'newer' is more or less
+ // recent.
+ if (older->previous_ghost != nullptr) {
+ if (CompareDuplicatedRows(*older->previous_ghost, *newer) > 0) {
+ // 'older' was more recent.
+ return SetPreviousGhost(older->previous_ghost, newer, must_copy /* clone */, arena);
+ }
+ // 'newer' was more recent.
+ if (must_copy) {
+ CompactionInputRow* newer_copy;
+ RETURN_NOT_OK(CloneCompactionInputRow(newer, &newer_copy, arena));
+ newer = newer_copy;
+ }
+ // 'older->previous_ghost' is already in 'arena' so avoid the copy.
+ RETURN_NOT_OK(SetPreviousGhost(newer,
+ older->previous_ghost,
+ false /* don't clone */,
+ arena));
+ older->previous_ghost = newer;
+ return Status::OK();
}
- // Duplicated rows have disjoint histories, we don't need to get the latest
- // mutation, the first one should be enough for the sake of determining the most recent
- // row, but in debug mode do get the latest to make sure one of the rows is a ghost.
- const Mutation* left_latest = left.redo_head;
- const Mutation* right_latest = right.redo_head;
- int ret = left_latest->timestamp().CompareTo(right_latest->timestamp());
-#ifndef NDEBUG
- AdvanceToLastInList(&left_latest);
- AdvanceToLastInList(&right_latest);
- int debug_ret = left_latest->timestamp().CompareTo(right_latest->timestamp());
- if (debug_ret != 0) {
- // If in fact both rows were deleted at the same time, this is OK -- we could
- // have a case like TestRandomAccess.TestFuzz3, in which a single batch
- // DELETED from the DRS, INSERTed into MRS, and DELETED from MRS. In that case,
- // the timestamp of the last REDO will be the same and we can pick whichever
- // we like.
- CHECK_EQ(ret, debug_ret);
+ if (must_copy) {
+ CompactionInputRow* newer_copy;
+ RETURN_NOT_OK(CloneCompactionInputRow(newer, &newer_copy, arena));
+ newer = newer_copy;
}
-#endif
- return ret;
+ older->previous_ghost = newer;
+ return Status::OK();
}
- static void AdvanceToLastInList(const Mutation** m) {
- const Mutation* next;
- while ((next = (*m)->acquire_next()) != nullptr) {
- *m = next;
+ // Duplicates are rare and allocating for the worst case (all the rows in all but one of
+ // the inputs are duplicates) is expensive, so we create RowBlocks on demand with just
+ // space for a few rows. If a row block is exhausted a new one is allocated.
+ RowBlockRow NewRow() {
+ rowid_t row_idx = num_dup_rows_ % kDuplicatedRowsPerBlock;
+ num_dup_rows_++;
+ if (row_idx == 0) {
+ duplicated_rows_.push_back(std::unique_ptr<RowBlock>(
+ new RowBlock(*schema_, kDuplicatedRowsPerBlock, static_cast<Arena*>(nullptr))));
}
+ return duplicated_rows_.back()->row(row_idx);
}
const Schema* schema_;
vector<MergeState *> states_;
Arena* prepared_block_arena_;
+
+ // Vector to keep blocks that store duplicated row data.
+ // This needs to be stored internally as row data for ghosts might have been deleted
+ // by the the time the most recent version row is processed.
+ vector<std::unique_ptr<RowBlock>> duplicated_rows_;
+ int num_dup_rows_;
+
+ enum {
+ kDuplicatedRowsPerBlock = 10
+ };
+
};
+// Advances 'head' while the timestamp of the 'next' mutation is bigger than or equal to 'ts'
+// and 'next' is not null.
+void AdvanceWhileNextEqualToOrBiggerThan(Mutation** head, Timestamp ts) {
+ while ((*head)->next() != nullptr && (*head)->next()->timestamp() >= ts) {
+ *head = (*head)->next();
+ }
+}
+
+// Merges two undo histories into one with decreasing timestamp order and returns the new head.
+Mutation* MergeUndoHistories(Mutation* left, Mutation* right) {
+
+ if (PREDICT_FALSE(left == nullptr)) {
+ return right;
+ }
+ if (PREDICT_FALSE(right == nullptr)) {
+ return left;
+ }
+
+ Mutation* head;
+ if (left->timestamp() >= right->timestamp()) {
+ head = left;
+ } else {
+ head = right;
+ }
+
+ while (left != nullptr && right != nullptr) {
+ if (left->timestamp() >= right->timestamp()) {
+ AdvanceWhileNextEqualToOrBiggerThan(&left, right->timestamp());
+ Mutation* next = left->next();
+ left->set_next(right);
+ left = next;
+ continue;
+ }
+ AdvanceWhileNextEqualToOrBiggerThan(&right, left->timestamp());
+ Mutation* next = right->next();
+ right->set_next(left);
+ right = next;
+ }
+ return head;
+}
+
+// If 'old_row' has previous versions, this transforms prior version in undos and adds them
+// to 'new_undo_head'.
+Status MergeDuplicatedRowHistory(CompactionInputRow* old_row,
+ Mutation** new_undo_head,
+ Arena* arena) {
+ if (PREDICT_TRUE(old_row->previous_ghost == nullptr)) return Status::OK();
+
+ // Use an all inclusive snapshot as all of the previous version's undos and redos
+ // are guaranteed to be committed, otherwise the compaction wouldn't be able to
+ // see the new row.
+ MvccSnapshot all_snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions();
+
+ faststring dst;
+
+ CompactionInputRow* previous_ghost = old_row->previous_ghost;
+ while (previous_ghost != nullptr) {
+
+ // First step is to transform the old rows REDO's into undos, if there are any.
+ // This simplifies this for several reasons:
+ // - will be left with most up-to-date version of the old row
+ // - only have one REDO left to deal with (the delete)
+ // - can reuse ApplyMutationsAndGenerateUndos()
+ Mutation* pv_new_undos_head = nullptr;
+ Mutation* pv_delete_redo = nullptr;
+
+ RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(all_snap,
+ *previous_ghost,
+ &pv_new_undos_head,
+ &pv_delete_redo,
+ arena,
+ &previous_ghost->row));
+
+ // We should be left with only one redo, the delete.
+ CHECK(pv_delete_redo != nullptr);
+ CHECK(pv_delete_redo->changelist().is_delete());
+ CHECK(pv_delete_redo->next() == nullptr);
+
+ // Now transform the redo delete into an undo (reinsert), which will contain the previous
+ // ghost. The reinsert will have the timestamp of the delete.
+ dst.clear();
+ RowChangeListEncoder undo_encoder(&dst);
+ undo_encoder.SetToReinsert(previous_ghost->row);
+ Mutation* pv_reinsert = Mutation::CreateInArena(arena,
+ pv_delete_redo->timestamp(),
+ undo_encoder.as_changelist());
+
+ // Make the reinsert point to the rest of the undos.
+ pv_reinsert->set_next(pv_new_undos_head);
+
+ // Merge the UNDO lists.
+ *new_undo_head = MergeUndoHistories(*new_undo_head, pv_reinsert);
+
+ // ... handle a previous ghost if there is any.
+ previous_ghost = previous_ghost->previous_ghost;
+ }
+ return Status::OK();
+}
+
// Makes the current head point to the 'new_head', if it's not null, and
// makes 'new_head' the new head.
void SetHead(Mutation** current_head, Mutation* new_head) {
@@ -569,7 +793,19 @@ string RowToString(const RowBlockRow& row, const Mutation* redo_head, const Muta
}
string CompactionInputRowToString(const CompactionInputRow& input_row) {
- return RowToString(input_row.row, input_row.redo_head, input_row.undo_head);
+ if (input_row.previous_ghost == nullptr) {
+ return RowToString(input_row.row, input_row.redo_head, input_row.undo_head);
+ }
+ string ret = RowToString(input_row.row, input_row.redo_head, input_row.undo_head);
+ const CompactionInputRow* previous_ghost = input_row.previous_ghost;
+ while (previous_ghost != nullptr) {
+ ret.append(" Previous Ghost: ");
+ ret.append(RowToString(previous_ghost->row,
+ previous_ghost->redo_head,
+ previous_ghost->undo_head));
+ previous_ghost = previous_ghost->previous_ghost;
+ }
+ return ret;
}
////////////////////////////////////////////////////////////
@@ -693,9 +929,6 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
Mutation** new_redo_head,
Arena* arena,
RowBlockRow* dst_row) {
-
- // At the time of writing, REINSERT is never encoded as an UNDO and the base
- // data can never be in a deleted state (see KUDU-237).
bool is_deleted = false;
#define ERROR_LOG_CONTEXT \
@@ -764,40 +997,44 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
undo_encoder.as_changelist());
break;
}
+ // When we see a reinsert REDO we do the following:
+ // 1 - Reset the REDO head, which contained a DELETE REDO.
+ // 2 - Apply the REINSERT to the row, passing an undo_encoder that encodes the state of
+ // the row prior to to the REINSERT.
+ // 3 - Create a mutation for the REINSERT above and add it to the UNDOs, this mutation
+ // will have the timestamp of the DELETE REDO.
+ // 4 - Create a new delete UNDO. This mutation will have the timestamp of the REINSERT REDO.
case RowChangeList::kReinsert: {
DCHECK(is_deleted) << "Got REINSERT for a non-deleted row. " << ERROR_LOG_CONTEXT;
CHECK(redo_delete != nullptr) << "Got REINSERT without a redo DELETE. "
- << ERROR_LOG_CONTEXT;
+ << ERROR_LOG_CONTEXT;
redo_decoder.TwiddleDeleteStatus(&is_deleted);
+ Timestamp delete_ts = redo_delete->timestamp();
+ // 1 - Reset the delete REDO.
+ redo_delete = nullptr;
- // Right now when a REINSERT mutation is found it is treated as a new insert and it
- // clears the whole row history before it. We apply the reinsert to the new
- // row but disregard the undo reinsert that is created.
- //
- // TODO(dralves) KUDU-237 Actually store the REINSERT
+ // 2 - Apply the changes of the reinsert to the latest version of the row
+ // capturing the old row while we're at it.
+ // TODO(dralves) Make Reinserts set defaults on the dest row. See KUDU-1760.
undo_encoder.SetToReinsert();
RETURN_NOT_OK_LOG(redo_decoder.MutateRowAndCaptureChanges(
dst_row, static_cast<Arena*>(nullptr), &undo_encoder), ERROR,
- "Unable to apply reinsert undo. \n " + ERROR_LOG_CONTEXT);
+ "Unable to apply reinsert undo. \n" + ERROR_LOG_CONTEXT);
- // Discard the REINSERT and store a DELETE instead.
+ // 3 - Create a mutation for the REINSERT above and add it to the UNDOs.
+ current_undo = Mutation::CreateInArena(arena,
+ delete_ts,
+ undo_encoder.as_changelist());
+ SetHead(&undo_head, current_undo);
+
+ // 4 - Create a DELETE mutation and add it to the UNDOs.
undo_encoder.Reset();
undo_encoder.SetToDelete();
-
- // Reset the UNDO head, losing all previous undos.
- undo_head = Mutation::CreateInArena(arena,
- redo_mut->timestamp(),
- undo_encoder.as_changelist());
-
- // Also reset the previous redo head since it stored the delete which was nullified
- // by this reinsert
- redo_delete = nullptr;
-
- if (PREDICT_FALSE(VLOG_IS_ON(2))) {
- VLOG(2) << "Found REINSERT REDO, cannot create UNDO for it, resetting row history "
- "under snapshot: " << snap.ToString() << ERROR_LOG_CONTEXT;
- }
+ current_undo = Mutation::CreateInArena(arena,
+ redo_mut->timestamp(),
+ undo_encoder.as_changelist());
+ SetHead(&undo_head, current_undo);
break;
}
default: LOG(FATAL) << "Unknown mutation type!" << ERROR_LOG_CONTEXT;
@@ -851,12 +1088,17 @@ Status FlushCompactionInput(CompactionInput* input,
input->PreparedBlockArena(),
&dst_row));
+ // Merge the histories of 'input_row' with previous ghosts, if there are any.
+ RETURN_NOT_OK(MergeDuplicatedRowHistory(input_row,
+ &new_undos_head,
+ input->PreparedBlockArena()));
+
// Remove ancient UNDOS and check whether the row should be garbage collected.
bool is_garbage_collected;
RemoveAncientUndos(history_gc_opts,
- &new_undos_head,
- new_redos_head,
- &is_garbage_collected);
+ &new_undos_head,
+ new_redos_head,
+ &is_garbage_collected);
DVLOG(4) << "Output Row: " << RowToString(dst_row, new_redos_head, new_undos_head) <<
"; Was garbage collected? " << is_garbage_collected;
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/compaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.h b/src/kudu/tablet/compaction.h
index a430688..bead82e 100644
--- a/src/kudu/tablet/compaction.h
+++ b/src/kudu/tablet/compaction.h
@@ -109,7 +109,8 @@ class CompactionInput {
// Returns the arena for this compaction input corresponding to the last
// prepared block. This must be called *after* PrepareBlock() as if this
// is a MergeCompactionInput only then will the right arena be selected.
- virtual Arena* PreparedBlockArena() = 0;
+ virtual Arena* PreparedBlockArena() = 0;
+
virtual Status FinishBlock() = 0;
virtual bool HasMoreBlocks() = 0;
@@ -162,6 +163,15 @@ struct CompactionInputRow {
Mutation* redo_head;
// The current undo head for this row, may be null if all undos were garbage collected.
Mutation* undo_head;
+
+ // When the same row is found in multiple rowsets because of ghost rows, this points
+ // to one that older in terms of row history.
+ CompactionInputRow* previous_ghost;
+
+ CompactionInputRow() :
+ redo_head(nullptr),
+ undo_head(nullptr),
+ previous_ghost(nullptr) {}
};
// Function shared by flushes and compactions. Removes UNDO Mutations
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/delta_store.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.cc b/src/kudu/tablet/delta_store.cc
index ba91f29..137bcda 100644
--- a/src/kudu/tablet/delta_store.cc
+++ b/src/kudu/tablet/delta_store.cc
@@ -38,7 +38,6 @@ string DeltaKeyAndUpdate::Stringify(DeltaType type, const Schema& schema, bool p
atoi(key.timestamp().ToString().c_str()))
: Substitute("$0@tx$1", key.row_idx(),
key.timestamp().ToString()))));
-
}
Status DebugDumpDeltaIterator(DeltaType type,
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/deltafile.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.cc b/src/kudu/tablet/deltafile.cc
index cd1f5be..1664133 100644
--- a/src/kudu/tablet/deltafile.cc
+++ b/src/kudu/tablet/deltafile.cc
@@ -100,12 +100,6 @@ Status DeltaFileWriter::FinishAndReleaseBlock(ScopedWritableBlockCloser* closer)
Status DeltaFileWriter::DoAppendDelta(const DeltaKey &key,
const RowChangeList &delta) {
Slice delta_slice(delta.slice());
-
- // See TODO in RowChangeListEncoder::SetToReinsert
- CHECK(!delta.is_reinsert())
- << "TODO: REINSERT deltas cannot currently be written to disk "
- << "since they don't have a standalone encoded form.";
-
tmp_buf_.clear();
// Write the encoded form of the key to the file.
@@ -639,14 +633,12 @@ struct ApplyingVisitor {
const Schema* schema = dfi->projection_;
RowChangeListDecoder decoder((RowChangeList(deltas)));
RETURN_NOT_OK(decoder.Init());
- if (decoder.is_update()) {
+ if (decoder.is_update() || decoder.is_reinsert()) {
return decoder.ApplyToOneColumn(rel_idx, dst, *schema, col_to_apply, dst->arena());
- } else if (decoder.is_delete()) {
- // If it's a DELETE, then it will be processed by DeletingVisitor.
- return Status::OK();
- } else {
- dfi->FatalUnexpectedDelta(key, deltas, "Expect only UPDATE or DELETE deltas on disk");
}
+
+ DCHECK(decoder.is_delete());
+ // If it's a DELETE, then it will be processed by LivenessVisitor.
return Status::OK();
}
@@ -686,16 +678,15 @@ Status DeltaFileIterator::ApplyUpdates(size_t col_to_apply, ColumnBlock *dst) {
DVLOG(3) << "Applying REDO mutations to " << col_to_apply;
ApplyingVisitor<REDO> visitor = {this, col_to_apply, dst};
return VisitMutations(&visitor);
- } else {
- DVLOG(3) << "Applying UNDO mutations to " << col_to_apply;
- ApplyingVisitor<UNDO> visitor = {this, col_to_apply, dst};
- return VisitMutations(&visitor);
}
+ DVLOG(3) << "Applying UNDO mutations to " << col_to_apply;
+ ApplyingVisitor<UNDO> visitor = {this, col_to_apply, dst};
+ return VisitMutations(&visitor);
}
-// Visitor which applies deletes to the selection vector.
+// Visitor which establishes the liveness of a row by applying deletes and reinserts.
template<DeltaType Type>
-struct DeletingVisitor {
+struct LivenessVisitor {
Status Visit(const DeltaKey &key, const Slice &deltas, bool* continue_visit);
@@ -710,12 +701,19 @@ struct DeletingVisitor {
// If this is an update the row must be selected.
DCHECK(sel_vec->IsRowSelected(rel_idx));
return Status::OK();
- } else if (decoder.is_delete()) {
+ }
+
+ if (decoder.is_delete()) {
DVLOG(3) << "Row deleted";
sel_vec->SetRowUnselected(rel_idx);
- } else {
- dfi->FatalUnexpectedDelta(key, deltas, "Expect only UPDATE or DELETE deltas on disk");
+ return Status::OK();
}
+
+ DCHECK(decoder.is_reinsert());
+ DVLOG(3) << "Re-selected the row (reinsert)";
+ // If this is a reinsert the row must be unselected.
+ DCHECK(!sel_vec->IsRowSelected(rel_idx));
+ sel_vec->SetRowSelected(rel_idx);
return Status::OK();
}
@@ -724,7 +722,7 @@ struct DeletingVisitor {
};
template<>
-inline Status DeletingVisitor<REDO>::Visit(const DeltaKey& key,
+inline Status LivenessVisitor<REDO>::Visit(const DeltaKey& key,
const Slice& deltas,
bool* continue_visit) {
if (IsRedoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) {
@@ -734,7 +732,7 @@ inline Status DeletingVisitor<REDO>::Visit(const DeltaKey& key,
}
template<>
-inline Status DeletingVisitor<UNDO>::Visit(const DeltaKey& key,
+inline Status LivenessVisitor<UNDO>::Visit(const DeltaKey& key,
const Slice& deltas, bool*
continue_visit) {
if (IsUndoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) {
@@ -748,13 +746,12 @@ Status DeltaFileIterator::ApplyDeletes(SelectionVector *sel_vec) {
DCHECK_LE(prepared_count_, sel_vec->nrows());
if (delta_type_ == REDO) {
DVLOG(3) << "Applying REDO deletes";
- DeletingVisitor<REDO> visitor = { this, sel_vec};
- return VisitMutations(&visitor);
- } else {
- DVLOG(3) << "Applying UNDO deletes";
- DeletingVisitor<UNDO> visitor = { this, sel_vec};
+ LivenessVisitor<REDO> visitor = { this, sel_vec };
return VisitMutations(&visitor);
}
+ DVLOG(3) << "Applying UNDO deletes";
+ LivenessVisitor<UNDO> visitor = { this, sel_vec };
+ return VisitMutations(&visitor);
}
// Visitor which, for each mutation, adds it into a ColumnBlock of
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/deltafile.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.h b/src/kudu/tablet/deltafile.h
index ea040d7..5706b08 100644
--- a/src/kudu/tablet/deltafile.h
+++ b/src/kudu/tablet/deltafile.h
@@ -53,7 +53,7 @@ struct ApplyingVisitor;
template<DeltaType Type>
struct CollectingVisitor;
template<DeltaType Type>
-struct DeletingVisitor;
+struct LivenessVisitor;
class DeltaFileWriter {
public:
@@ -208,8 +208,8 @@ class DeltaFileIterator : public DeltaIterator {
friend struct ApplyingVisitor<UNDO>;
friend struct CollectingVisitor<REDO>;
friend struct CollectingVisitor<UNDO>;
- friend struct DeletingVisitor<REDO>;
- friend struct DeletingVisitor<UNDO>;
+ friend struct LivenessVisitor<REDO>;
+ friend struct LivenessVisitor<UNDO>;
friend struct FilterAndAppendVisitor;
DISALLOW_COPY_AND_ASSIGN(DeltaFileIterator);
http://git-wip-us.apache.org/repos/asf/kudu/blob/abab2ced/src/kudu/tablet/tablet-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test.cc b/src/kudu/tablet/tablet-test.cc
index 90bc5ef..1de12bd 100644
--- a/src/kudu/tablet/tablet-test.cc
+++ b/src/kudu/tablet/tablet-test.cc
@@ -219,6 +219,69 @@ TYPED_TEST(TestTablet, TestInsertDuplicateKey) {
ASSERT_EQ(1, this->TabletCount());
}
+// Tests that we are able to handle reinserts properly.
+//
+// Namely tests that:
+// - We're able to perform multiple reinserts in a MRS, flush them
+// and that all versions of the row are still visible.
+// - After we've flushed the reinserts above, we can perform a
+// new reinsert in a new MRS, flush that MRS and compact the row
+// DRS together, all while preserving the full row history.
+TYPED_TEST(TestTablet, TestReinserts) {
+ LocalTabletWriter writer(this->tablet().get(), &this->client_schema_);
+
+ vector<MvccSnapshot> snaps;
+ // In the first snap there's no row.
+ snaps.push_back(MvccSnapshot(*this->tablet()->mvcc_manager()));
+
+ // Insert one row.
+ ASSERT_OK(this->InsertTestRow(&writer, 1, 0));
+
+ // In the second snap the row exists and has value 0.
+ snaps.push_back(MvccSnapshot(*this->tablet()->mvcc_manager()));
+
+ // Now delete the test row.
+ ASSERT_OK(this->DeleteTestRow(&writer, 1));
+
+ // In the third snap the row doesn't exist.
+ snaps.push_back(MvccSnapshot(*this->tablet()->mvcc_manager()));
+
+ // Reinsert the row.
+ ASSERT_OK(this->InsertTestRow(&writer, 1, 1));
+
+ // In the fourth snap the row exists again and has value 1.
+ snaps.push_back(MvccSnapshot(*this->tablet()->mvcc_manager()));
+
+ // .. and delete the row again.
+ ASSERT_OK(this->DeleteTestRow(&writer, 1));
+
+ // In the fifth snap the row has been deleted.
+ snaps.push_back(MvccSnapshot(*this->tablet()->mvcc_manager()));
+
+ // Now flush the MRS all versions of the tablet should be visible,
+ // depending on the chosen snapshot.
+ ASSERT_OK(this->tablet()->Flush());
+
+ vector<vector<string>* > expected_rows;
+ CollectRowsForSnapshots(this->tablet().get(), this->client_schema_,
+ snaps, &expected_rows);
+
+ ASSERT_EQ(expected_rows.size(), 5);
+ ASSERT_EQ(expected_rows[0]->size(), 0) << "Got the wrong result from snap: "
+ << snaps[0].ToString();
+ ASSERT_EQ(expected_rows[1]->size(), 1) << "Got the wrong result from snap: "
+ << snaps[1].ToString();
+ ASSERT_STR_CONTAINS((*expected_rows[1])[0], "int32 key_idx=1, int32 val=0)");
+ ASSERT_EQ(expected_rows[2]->size(), 0) << "Got the wrong result from snap: "
+ << snaps[2].ToString();
+ ASSERT_EQ(expected_rows[3]->size(), 1) << "Got the wrong result from snap: "
+ << snaps[3].ToString();
+ ASSERT_STR_CONTAINS((*expected_rows[3])[0], "int32 key_idx=1, int32 val=1)");
+ ASSERT_EQ(expected_rows[4]->size(), 0) << "Got the wrong result from snap: "
+ << snaps[4].ToString();
+
+ STLDeleteElements(&expected_rows);
+}
// Test flushes and compactions dealing with deleted rows.
TYPED_TEST(TestTablet, TestDeleteWithFlushAndCompact) {
[3/5] kudu git commit: Add snapshot scans to fuzz-itest
Posted by to...@apache.org.
Add snapshot scans to fuzz-itest
This adds a new operation to fuzz-itest: snapshot scans
at a timestamp.
When generating random operations, scans at a timestamp are
now in the mix. The generator records how timestamps will
progress with writes and is careful to make sure that scans
happen in the past. After the test case is generated, but
before it is run, the timestamps for the scans are deduplicated
and sorted to that we save the state immediately before those
(and only those) timestamps.
This would fail very often before the REINSERT patches but
passes with it.
Ran 500 loops of fuzz-itest in dist-test in ASAN with:
- KUDU_ALLOW_SLOW_TESTS=1
- --stress_cpu_threads=4
All tests passed. Result:
http://dist-test.cloudera.org//job?job_id=david.alves.1480096588.30319
Change-Id: I4d15129e83c5c9afa9b643e674c8a23e18a2fa0e
Reviewed-on: http://gerrit.cloudera.org:8080/4996
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/10f2ce20
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/10f2ce20
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/10f2ce20
Branch: refs/heads/master
Commit: 10f2ce208fd646c6d2db0f8277ab0342b52cd080
Parents: abab2ce
Author: David Alves <dr...@apache.org>
Authored: Mon Nov 7 21:13:27 2016 -0800
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Wed Nov 30 10:14:38 2016 +0000
----------------------------------------------------------------------
src/kudu/integration-tests/fuzz-itest.cc | 220 +++++++++++++++++++++-----
src/kudu/tablet/key_value_test_schema.h | 21 ++-
src/kudu/tserver/tablet_service.cc | 13 +-
3 files changed, 206 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/10f2ce20/src/kudu/integration-tests/fuzz-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/fuzz-itest.cc b/src/kudu/integration-tests/fuzz-itest.cc
index f4a8c8d..00cd2ec 100644
--- a/src/kudu/integration-tests/fuzz-itest.cc
+++ b/src/kudu/integration-tests/fuzz-itest.cc
@@ -21,6 +21,7 @@
#include <glog/stl_logging.h>
#include <gtest/gtest.h>
#include <gflags/gflags.h>
+#include <list>
#include <string>
#include <vector>
@@ -33,6 +34,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/master/mini_master.h"
#include "kudu/integration-tests/mini_cluster.h"
+#include "kudu/server/logical_clock.h"
#include "kudu/tablet/key_value_test_schema.h"
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet_peer.h"
@@ -44,13 +46,9 @@
DEFINE_int32(keyspace_size, 2, "number of distinct primary keys to test with");
DECLARE_bool(enable_maintenance_manager);
+DECLARE_bool(scanner_allow_snapshot_scans_with_logical_timestamps);
DECLARE_bool(use_hybrid_clock);
-using boost::optional;
-using std::string;
-using std::vector;
-using std::unique_ptr;
-
// The type of operation in a sequence of operations generated by
// the fuzz test.
enum TestOpType {
@@ -66,6 +64,7 @@ enum TestOpType {
TEST_MAJOR_COMPACT_DELTAS,
TEST_COMPACT_TABLET,
TEST_RESTART_TS,
+ TEST_SCAN_AT_TIMESTAMP,
TEST_NUM_OP_TYPES // max value for enum
};
MAKE_ENUM_LIMITS(TestOpType, TEST_INSERT, TEST_NUM_OP_TYPES);
@@ -74,11 +73,10 @@ const char* kTableName = "table";
namespace kudu {
+using boost::optional;
using client::KuduClient;
using client::KuduClientBuilder;
-using client::KuduColumnSchema;
using client::KuduDelete;
-using client::KuduInsert;
using client::KuduPredicate;
using client::KuduScanBatch;
using client::KuduScanner;
@@ -92,6 +90,12 @@ using client::KuduUpsert;
using client::KuduValue;
using client::KuduWriteOperation;
using client::sp::shared_ptr;
+using std::list;
+using std::map;
+using std::string;
+using std::vector;
+using std::unique_ptr;
+using strings::Substitute;
namespace tablet {
@@ -107,7 +111,8 @@ const char* TestOpType_names[] = {
"TEST_MINOR_COMPACT_DELTAS",
"TEST_MAJOR_COMPACT_DELTAS",
"TEST_COMPACT_TABLET",
- "TEST_RESTART_TS"
+ "TEST_RESTART_TS",
+ "TEST_SCAN_AT_TIMESTAMP"
};
// An operation in a fuzz-test sequence.
@@ -115,11 +120,13 @@ struct TestOp {
// The op to run.
TestOpType type;
- // For INSERT/UPSERT/UPDATE/DELETE, the key of the row to be modified. Otherwise, unused.
- int row_key;
+ // For INSERT/UPSERT/UPDATE/DELETE, the key of the row to be modified.
+ // For SCAN_AT_TIMESTAMP the timestamp of the scan.
+ // Otherwise, unused.
+ int val;
string ToString() const {
- return strings::Substitute("{$0, $1}", TestOpType_names[type], row_key);
+ return strings::Substitute("{$0, $1}", TestOpType_names[type], val);
}
};
@@ -134,6 +141,7 @@ class FuzzTest : public KuduTest {
FuzzTest() {
FLAGS_enable_maintenance_manager = false;
FLAGS_use_hybrid_clock = false;
+ FLAGS_scanner_allow_snapshot_scans_with_logical_timestamps = true;
schema_ = client::KuduSchemaFromSchema(CreateKeyValueTestSchema());
}
@@ -277,8 +285,98 @@ class FuzzTest : public KuduTest {
return boost::none;
}
+ // Checks that the rows in 'found' match the state we've stored 'saved_values_' corresponding
+ // to 'timestamp'. 'errors' collects the errors found. If 'errors' is not empty it means the
+ // check failed.
+ void CheckRowsMatchAtTimestamp(int timestamp,
+ vector<ExpectedKeyValueRow> rows_found,
+ list<string>* errors) {
+ int saved_timestamp = -1;
+ auto iter = saved_values_.upper_bound(timestamp);
+ if (iter == saved_values_.end()) {
+ if (!rows_found.empty()) {
+ for (auto& found_row : rows_found) {
+ errors->push_back(Substitute("Found unexpected row: $0", found_row.ToString()));
+ }
+ }
+ } else {
+ saved_timestamp = iter->first;
+ const auto& saved = (*iter).second;
+ int found_idx = 0;
+ int expected_values_counter = 0;
+ for (auto& expected : saved) {
+ if (expected) {
+ expected_values_counter++;
+ ExpectedKeyValueRow expected_val = expected.value();
+ if (found_idx >= rows_found.size()) {
+ errors->push_back(Substitute("Didn't find expected value: $0",
+ expected_val.ToString()));
+ break;
+ }
+ ExpectedKeyValueRow found_val = rows_found[found_idx++];
+ if (expected_val.key != found_val.key) {
+ errors->push_back(Substitute("Mismached key. Expected: $0 Found: $1",
+ expected_val.ToString(), found_val.ToString()));
+ continue;
+ }
+ if (expected_val.val != found_val.val) {
+ errors->push_back(Substitute("Mismached value. Expected: $0 Found: $1",
+ expected_val.ToString(), found_val.ToString()));
+ continue;
+ }
+ }
+ }
+ if (rows_found.size() != expected_values_counter) {
+ errors->push_back(Substitute("Mismatched size. Expected: $0 rows. Found: $1 rows.",
+ expected_values_counter, rows_found.size()));
+ for (auto& found_row : rows_found) {
+ errors->push_back(Substitute("Found unexpected row: $0", found_row.ToString()));
+ }
+ }
+ }
+ if (!errors->empty()) {
+ errors->push_front(Substitute("Found errors while comparing a snapshot scan at $0 with the "
+ "values saved at $1. Errors:",
+ timestamp, saved_timestamp));
+ }
+ }
+
+ // Scan the tablet at 'timestamp' and compare the result to the saved values.
+ void CheckScanAtTimestamp(int timestamp) {
+ KuduScanner s(table_.get());
+ ASSERT_OK(s.SetReadMode(KuduScanner::ReadMode::READ_AT_SNAPSHOT));
+ ASSERT_OK(s.SetSnapshotRaw(timestamp));
+ ASSERT_OK(s.SetOrderMode(KuduScanner::OrderMode::ORDERED));
+ ASSERT_OK(s.Open());
+ vector<ExpectedKeyValueRow> found;
+ while (s.HasMoreRows()) {
+ KuduScanBatch batch;
+ ASSERT_OK(s.NextBatch(&batch));
+ for (KuduScanBatch::RowPtr row : batch) {
+ ExpectedKeyValueRow ret;
+ ASSERT_OK(row.GetInt32(0, &ret.key));
+ if (!row.IsNull(1)) {
+ ret.val = 0;
+ ASSERT_OK(row.GetInt32(1, ret.val.get_ptr()));
+ }
+ found.push_back(ret);
+ }
+ }
+
+ list<string> errors;
+ CheckRowsMatchAtTimestamp(timestamp, std::move(found), &errors);
+
+ string final_error;
+ if (!errors.empty()) {
+ for (const string& error : errors) {
+ final_error.append("\n" + error);
+ }
+ FAIL() << final_error;
+ }
+ }
+
protected:
- void RunFuzzCase(const vector<TestOp>& ops,
+ void RunFuzzCase(const vector<TestOp>& test_ops,
int update_multiplier);
KuduSchema schema_;
@@ -287,6 +385,10 @@ class FuzzTest : public KuduTest {
shared_ptr<KuduSession> session_;
shared_ptr<KuduTable> table_;
+ map<int,
+ vector<optional<ExpectedKeyValueRow>>,
+ std::greater<int>> saved_values_;
+
scoped_refptr<TabletPeer> tablet_peer_;
};
@@ -294,6 +396,7 @@ class FuzzTest : public KuduTest {
// fuzz test.
void GenerateTestCase(vector<TestOp>* ops, int len) {
vector<bool> exists(FLAGS_keyspace_size);
+ int op_timestamps = 0;
bool ops_pending = false;
bool data_in_mrs = false;
bool worth_compacting = false;
@@ -346,6 +449,7 @@ void GenerateTestCase(vector<TestOp>* ops, int len) {
if (ops_pending) {
ops->push_back({TEST_FLUSH_OPS, 0});
ops_pending = false;
+ op_timestamps++;
}
break;
case TEST_FLUSH_TABLET:
@@ -388,6 +492,14 @@ void GenerateTestCase(vector<TestOp>* ops, int len) {
case TEST_RESTART_TS:
ops->push_back({TEST_RESTART_TS, 0});
break;
+ case TEST_SCAN_AT_TIMESTAMP: {
+ int timestamp = 1;
+ if (op_timestamps > 0) {
+ timestamp = (rand() % op_timestamps) + 1;
+ }
+ ops->push_back({TEST_SCAN_AT_TIMESTAMP, timestamp});
+ break;
+ }
default:
LOG(FATAL);
}
@@ -409,35 +521,66 @@ void FuzzTest::RunFuzzCase(const vector<TestOp>& test_ops,
// into a test method in order to reproduce a failure.
LOG(INFO) << "test case:\n" << DumpTestCase(test_ops);
+ // Keep the vector of timestamps we'll scan at so that we save the expected state at those times.
+ vector<int> timestamps_to_scan;
+ for (const TestOp& test_op : test_ops) {
+ if (test_op.type == TEST_SCAN_AT_TIMESTAMP) {
+ timestamps_to_scan.push_back(test_op.val);
+ }
+ }
+ // Sort the scan timestamps in reverse order so that we can keep popping from the back and remove
+ // duplicates.
+ sort(timestamps_to_scan.begin(), timestamps_to_scan.end(), std::greater<int>());
+ timestamps_to_scan.erase(unique(timestamps_to_scan.begin(),
+ timestamps_to_scan.end()),
+ timestamps_to_scan.end() );
+
vector<optional<ExpectedKeyValueRow>> cur_val(FLAGS_keyspace_size);
vector<optional<ExpectedKeyValueRow>> pending_val(FLAGS_keyspace_size);
int i = 0;
for (const TestOp& test_op : test_ops) {
- optional<ExpectedKeyValueRow> val_in_table = GetRow(test_op.row_key);
- EXPECT_EQ(cur_val[test_op.row_key], val_in_table);
+ switch (test_op.type) {
+ case TEST_INSERT:
+ case TEST_UPSERT:
+ case TEST_UPSERT_PK_ONLY:
+ case TEST_UPDATE:
+ case TEST_DELETE:
+ EXPECT_EQ(cur_val[test_op.val], GetRow(test_op.val));
+ break;
+ default: break;
+ }
LOG(INFO) << test_op.ToString();
switch (test_op.type) {
case TEST_INSERT:
case TEST_UPSERT:
case TEST_UPSERT_PK_ONLY: {
- pending_val[test_op.row_key] = InsertOrUpsertRow(
- test_op.row_key, i++, pending_val[test_op.row_key], test_op.type);
+ pending_val[test_op.val] = InsertOrUpsertRow(
+ test_op.val, i++, pending_val[test_op.val], test_op.type);
break;
}
case TEST_UPDATE:
for (int j = 0; j < update_multiplier; j++) {
- pending_val[test_op.row_key] = MutateRow(test_op.row_key, i++);
+ pending_val[test_op.val] = MutateRow(test_op.val, i++);
}
break;
case TEST_DELETE:
- pending_val[test_op.row_key] = DeleteRow(test_op.row_key);
+ pending_val[test_op.val] = DeleteRow(test_op.val);
break;
- case TEST_FLUSH_OPS:
+ case TEST_FLUSH_OPS: {
FlushSessionOrDie(session_);
cur_val = pending_val;
+ int current_time = down_cast<kudu::server::LogicalClock*>(
+ tablet()->clock().get())->GetCurrentTime();
+ // Check if the next snapshot scan has a time that is higher than the current time.
+ // If it is, then store the state so that we can match it later to the scanned state.
+ if (!timestamps_to_scan.empty() && current_time >= timestamps_to_scan.back()) {
+ saved_values_[current_time] = cur_val;
+ timestamps_to_scan.pop_back();
+ }
break;
+ }
case TEST_FLUSH_TABLET:
ASSERT_OK(tablet()->Flush());
break;
@@ -456,17 +599,19 @@ void FuzzTest::RunFuzzCase(const vector<TestOp>& test_ops,
case TEST_RESTART_TS:
NO_FATALS(RestartTabletServer());
break;
+ case TEST_SCAN_AT_TIMESTAMP:
+ NO_FATALS(CheckScanAtTimestamp(test_op.val));
+ break;
default:
LOG(FATAL) << test_op.type;
}
}
}
-
// Generates a random test sequence and runs it.
// The logs of this test are designed to easily be copy-pasted and create
// more specific test cases like TestFuzz<N> below.
-TEST_F(FuzzTest, TestFuzz) {
+TEST_F(FuzzTest, TestRandomFuzz) {
SeedRandom();
vector<TestOp> test_ops;
GenerateTestCase(&test_ops, AllowSlowTests() ? 1000 : 50);
@@ -476,7 +621,7 @@ TEST_F(FuzzTest, TestFuzz) {
// Generates a random test case, but the UPDATEs are all repeated many times.
// This results in very large batches which are likely to span multiple delta blocks
// when flushed.
-TEST_F(FuzzTest, TestFuzzHugeBatches) {
+TEST_F(FuzzTest, TestRandomFuzzHugeBatches) {
SeedRandom();
vector<TestOp> test_ops;
GenerateTestCase(&test_ops, AllowSlowTests() ? 500 : 50);
@@ -490,27 +635,26 @@ TEST_F(FuzzTest, TestFuzzHugeBatches) {
RunFuzzCase(test_ops, update_multiplier);
}
-// A particular test case which previously failed TestFuzz.
TEST_F(FuzzTest, TestFuzz1) {
vector<TestOp> test_ops = {
- // Get an inserted row in a DRS.
- {TEST_INSERT, 0},
- {TEST_FLUSH_OPS, 0},
- {TEST_FLUSH_TABLET, 0},
+ // Get an inserted row in a DRS.
+ {TEST_INSERT, 0},
+ {TEST_FLUSH_OPS, 0},
+ {TEST_FLUSH_TABLET, 0},
- // DELETE in DMS, INSERT in MRS and flush again.
- {TEST_DELETE, 0},
- {TEST_INSERT, 0},
- {TEST_FLUSH_OPS, 0},
- {TEST_FLUSH_TABLET, 0},
+ // DELETE in DMS, INSERT in MRS and flush again.
+ {TEST_DELETE, 0},
+ {TEST_INSERT, 0},
+ {TEST_FLUSH_OPS, 0},
+ {TEST_FLUSH_TABLET, 0},
- // State:
- // RowSet RowSet(0):
- // (int32 key=1, int32 val=NULL) Undos: [@1(DELETE)] Redos (in DMS): [@2 DELETE]
- // RowSet RowSet(1):
- // (int32 key=1, int32 val=NULL) Undos: [@2(DELETE)] Redos: []
+ // State:
+ // RowSet RowSet(0):
+ // (int32 key=1, int32 val=NULL) Undos: [@1(DELETE)] Redos (in DMS): [@2 DELETE]
+ // RowSet RowSet(1):
+ // (int32 key=1, int32 val=NULL) Undos: [@2(DELETE)] Redos: []
- {TEST_COMPACT_TABLET, 0},
+ {TEST_COMPACT_TABLET, 0},
};
RunFuzzCase(test_ops);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/10f2ce20/src/kudu/tablet/key_value_test_schema.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/key_value_test_schema.h b/src/kudu/tablet/key_value_test_schema.h
index e81a4f2..479be39 100644
--- a/src/kudu/tablet/key_value_test_schema.h
+++ b/src/kudu/tablet/key_value_test_schema.h
@@ -21,10 +21,11 @@
#pragma once
#include <boost/optional.hpp>
-#include <boost/optional/optional_io.hpp>
#include <iostream>
+#include <string>
#include "kudu/common/schema.h"
+#include "kudu/gutil/strings/substitute.h"
namespace kudu {
@@ -40,6 +41,16 @@ struct ExpectedKeyValueRow {
bool operator==(const ExpectedKeyValueRow& other) const {
return key == other.key && val == other.val;
}
+
+ string ToString() const {
+ string ret = strings::Substitute("{$0,", key);
+ if (val == boost::none) {
+ ret.append("NULL}");
+ } else {
+ ret.append(strings::Substitute("$0}", *val));
+ }
+ return ret;
+ }
};
inline Schema CreateKeyValueTestSchema() {
@@ -48,13 +59,7 @@ inline Schema CreateKeyValueTestSchema() {
}
inline std::ostream& operator<<(std::ostream& o, const ExpectedKeyValueRow& t) {
- o << "{" << t.key << ", ";
- if (t.val == boost::none) {
- o << "NULL";
- } else {
- o << *t.val;
- }
- return o << "}";
+ return o << t.ToString();
}
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/10f2ce20/src/kudu/tserver/tablet_service.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index e0b334f..5177a5a 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -75,6 +75,10 @@ DEFINE_int32(scanner_batch_size_rows, 100,
TAG_FLAG(scanner_batch_size_rows, advanced);
TAG_FLAG(scanner_batch_size_rows, runtime);
+DEFINE_bool(scanner_allow_snapshot_scans_with_logical_timestamps, false,
+ "If set, the server will support snapshot scans with logical timestamps.");
+TAG_FLAG(scanner_allow_snapshot_scans_with_logical_timestamps, unsafe);
+
// Fault injection flags.
DEFINE_int32(scanner_inject_latency_on_each_batch_ms, 0,
"If set, the scanner will pause the specified number of milliesconds "
@@ -1747,13 +1751,18 @@ Status TabletServiceImpl::HandleScanAtSnapshot(const NewScanRequestPB& scan_pb,
// ... else we use the client provided one, but make sure it is not too far
// in the future as to be invalid.
} else {
- tmp_snap_timestamp.FromUint64(scan_pb.snap_timestamp());
+
Timestamp max_allowed_ts;
Status s = server_->clock()->GetGlobalLatest(&max_allowed_ts);
- if (!s.ok()) {
+ if (s.IsNotSupported() &&
+ PREDICT_TRUE(!FLAGS_scanner_allow_snapshot_scans_with_logical_timestamps)) {
return Status::NotSupported("Snapshot scans not supported on this server",
s.ToString());
}
+ tmp_snap_timestamp.FromUint64(scan_pb.snap_timestamp());
+
+ // Note: if 'max_allowed_ts' is not obtained from clock_->GetGlobalLatest() it's guaranteed
+ // to be higher than 'tmp_snap_timestamp'.
if (tmp_snap_timestamp > max_allowed_ts) {
return Status::InvalidArgument(
Substitute("Snapshot time $0 in the future. Max allowed timestamp is $1",
[5/5] kudu git commit: cfile: make encoder/decoder classes final
Posted by to...@apache.org.
cfile: make encoder/decoder classes final
Marking the classes final means that the compiler can be sure that a
virtual method call from within the class will always be satisfied by
the implementation in that same class (since no subclass could exist).
This increases inlining opportunities and other downstream
optimizations, and should improve performance a few percent in various
spots.
I verified the effect by running:
perf record ./build/latest/bin/cfile-test --gtest_filter=\*100MFileStringsPlain\*0
perf report -n | grep BinaryPlainBlockBuilder | c++filt
Before:
Overhead Samples Command Shared Object Symbol
6.22% 1943 cfile-test cfile-test [.] kudu::cfile::BinaryPlainBlockBuilder::Add(unsigned char const*, unsigned long)
1.23% 386 cfile-test cfile-test [.] kudu::cfile::BinaryPlainBlockBuilder::IsBlockFull() const
0.23% 73 cfile-test cfile-test [.] kudu::cfile::BinaryPlainBlockBuilder::Finish(unsigned int)
0.00% 1 cfile-test cfile-test [.] kudu::cfile::BinaryPlainBlockBuilder::Reset()
After:
Overhead Samples Command Shared Object Symbol
6.21% 1868 cfile-test cfile-test [.] kudu::cfile::BinaryPlainBlockBuilder::Add(unsigned char const*, unsigned long)
0.23% 69 cfile-test cfile-test [.] kudu::cfile::BinaryPlainBlockBuilder::Finish(unsigned int)
You can see that several functions were now inlined where they previously could
not be, and the sample count for 'Add' was also reduced.
Change-Id: Ia0b8337868330b43595a275e837a448e3ba0c066
Reviewed-on: http://gerrit.cloudera.org:8080/5202
Reviewed-by: Dan Burkert <da...@apache.org>
Tested-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/eb4b46b8
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/eb4b46b8
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/eb4b46b8
Branch: refs/heads/master
Commit: eb4b46b800a92d4964e5d0eaec4a4fd011463901
Parents: 6f42330
Author: Todd Lipcon <to...@apache.org>
Authored: Tue Nov 29 17:34:30 2016 -0800
Committer: Todd Lipcon <to...@apache.org>
Committed: Wed Nov 30 17:06:35 2016 +0000
----------------------------------------------------------------------
src/kudu/cfile/binary_dict_block.h | 4 ++--
src/kudu/cfile/binary_plain_block.h | 4 ++--
src/kudu/cfile/binary_prefix_block.cc | 5 +----
src/kudu/cfile/binary_prefix_block.h | 4 ++--
src/kudu/cfile/bshuf_block.h | 4 ++--
src/kudu/cfile/gvint_block.h | 4 ++--
src/kudu/cfile/plain_bitmap_block.h | 4 ++--
src/kudu/cfile/plain_block.h | 4 ++--
src/kudu/cfile/rle_block.h | 8 ++++----
9 files changed, 19 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/binary_dict_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/binary_dict_block.h b/src/kudu/cfile/binary_dict_block.h
index 95ff64c..c06a506 100644
--- a/src/kudu/cfile/binary_dict_block.h
+++ b/src/kudu/cfile/binary_dict_block.h
@@ -64,7 +64,7 @@ enum DictEncodingMode {
DictEncodingMode_max = 2
};
-class BinaryDictBlockBuilder : public BlockBuilder {
+class BinaryDictBlockBuilder final : public BlockBuilder {
public:
explicit BinaryDictBlockBuilder(const WriterOptions* options);
@@ -119,7 +119,7 @@ class BinaryDictBlockBuilder : public BlockBuilder {
class CFileIterator;
-class BinaryDictBlockDecoder : public BlockDecoder {
+class BinaryDictBlockDecoder final : public BlockDecoder {
public:
explicit BinaryDictBlockDecoder(Slice slice, CFileIterator* iter);
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/binary_plain_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/binary_plain_block.h b/src/kudu/cfile/binary_plain_block.h
index 1004596..8c9cd49 100644
--- a/src/kudu/cfile/binary_plain_block.h
+++ b/src/kudu/cfile/binary_plain_block.h
@@ -39,7 +39,7 @@ namespace cfile {
struct WriterOptions;
-class BinaryPlainBlockBuilder : public BlockBuilder {
+class BinaryPlainBlockBuilder final : public BlockBuilder {
public:
explicit BinaryPlainBlockBuilder(const WriterOptions *options);
@@ -88,7 +88,7 @@ class BinaryPlainBlockBuilder : public BlockBuilder {
};
-class BinaryPlainBlockDecoder : public BlockDecoder {
+class BinaryPlainBlockDecoder final : public BlockDecoder {
public:
explicit BinaryPlainBlockDecoder(Slice slice);
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/binary_prefix_block.cc
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/binary_prefix_block.cc b/src/kudu/cfile/binary_prefix_block.cc
index 0c299b0..faff8d9 100644
--- a/src/kudu/cfile/binary_prefix_block.cc
+++ b/src/kudu/cfile/binary_prefix_block.cc
@@ -130,10 +130,7 @@ int BinaryPrefixBlockBuilder::Add(const uint8_t *vals, size_t count) {
int added = 0;
const Slice* slices = reinterpret_cast<const Slice*>(vals);
Slice prev_val(last_val_);
- // We generate a static call to IsBlockFull() to avoid the vtable lookup
- // in this hot path.
- while (!BinaryPrefixBlockBuilder::IsBlockFull() &&
- added < count) {
+ while (!IsBlockFull() && added < count) {
const Slice val = slices[added];
int old_size = buffer_.size();
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/binary_prefix_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/binary_prefix_block.h b/src/kudu/cfile/binary_prefix_block.h
index 3094b07..a330a09 100644
--- a/src/kudu/cfile/binary_prefix_block.h
+++ b/src/kudu/cfile/binary_prefix_block.h
@@ -33,7 +33,7 @@ struct WriterOptions;
// Encoding for data blocks of binary data that have common prefixes.
// This encodes in a manner similar to LevelDB (prefix coding)
-class BinaryPrefixBlockBuilder : public BlockBuilder {
+class BinaryPrefixBlockBuilder final : public BlockBuilder {
public:
explicit BinaryPrefixBlockBuilder(const WriterOptions *options);
@@ -82,7 +82,7 @@ class BinaryPrefixBlockBuilder : public BlockBuilder {
};
// Decoder for BINARY type, PREFIX encoding
-class BinaryPrefixBlockDecoder : public BlockDecoder {
+class BinaryPrefixBlockDecoder final : public BlockDecoder {
public:
explicit BinaryPrefixBlockDecoder(Slice slice);
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/bshuf_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/bshuf_block.h b/src/kudu/cfile/bshuf_block.h
index 794da45..7ec0bae 100644
--- a/src/kudu/cfile/bshuf_block.h
+++ b/src/kudu/cfile/bshuf_block.h
@@ -82,7 +82,7 @@ void AbortWithBitShuffleError(int64_t val) ATTRIBUTE_NORETURN;
// The header is followed by the bitshuffle-compressed element data.
//
template<DataType Type>
-class BShufBlockBuilder : public BlockBuilder {
+class BShufBlockBuilder final : public BlockBuilder {
public:
explicit BShufBlockBuilder(const WriterOptions* options)
: count_(0),
@@ -215,7 +215,7 @@ template<>
Slice BShufBlockBuilder<UINT32>::Finish(rowid_t ordinal_pos);
template<DataType Type>
-class BShufBlockDecoder : public BlockDecoder {
+class BShufBlockDecoder final : public BlockDecoder {
public:
explicit BShufBlockDecoder(Slice slice)
: data_(std::move(slice)),
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/gvint_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/gvint_block.h b/src/kudu/cfile/gvint_block.h
index 6c386fd..d8d3042 100644
--- a/src/kudu/cfile/gvint_block.h
+++ b/src/kudu/cfile/gvint_block.h
@@ -43,7 +43,7 @@ using std::vector;
//
// See AppendGroupVarInt32(...) for details on the varint
// encoding.
-class GVIntBlockBuilder : public BlockBuilder {
+class GVIntBlockBuilder final : public BlockBuilder {
public:
explicit GVIntBlockBuilder(const WriterOptions *options);
@@ -98,7 +98,7 @@ class GVIntBlockBuilder : public BlockBuilder {
};
// Decoder for UINT32 type, GROUP_VARINT coding
-class GVIntBlockDecoder : public BlockDecoder {
+class GVIntBlockDecoder final : public BlockDecoder {
public:
explicit GVIntBlockDecoder(Slice slice);
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/plain_bitmap_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/plain_bitmap_block.h b/src/kudu/cfile/plain_bitmap_block.h
index 04e1f89..5414344 100644
--- a/src/kudu/cfile/plain_bitmap_block.h
+++ b/src/kudu/cfile/plain_bitmap_block.h
@@ -41,7 +41,7 @@ struct WriterOptions;
// A plain encoder for the BOOL datatype: stores a column of BOOL values
// as a packed bitmap.
//
-class PlainBitMapBlockBuilder : public BlockBuilder {
+class PlainBitMapBlockBuilder final : public BlockBuilder {
public:
explicit PlainBitMapBlockBuilder(const WriterOptions* options)
: writer_(&buf_),
@@ -106,7 +106,7 @@ class PlainBitMapBlockBuilder : public BlockBuilder {
//
// Plain decoder for the BOOL datatype
//
-class PlainBitMapBlockDecoder : public BlockDecoder {
+class PlainBitMapBlockDecoder final : public BlockDecoder {
public:
explicit PlainBitMapBlockDecoder(Slice slice)
: data_(std::move(slice)),
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/plain_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/plain_block.h b/src/kudu/cfile/plain_block.h
index e01f3fa..4f9bcd6 100644
--- a/src/kudu/cfile/plain_block.h
+++ b/src/kudu/cfile/plain_block.h
@@ -43,7 +43,7 @@ static const size_t kPlainBlockHeaderSize = sizeof(uint32_t) * 2;
// A plain encoder for generic fixed size data types.
//
template<DataType Type>
-class PlainBlockBuilder : public BlockBuilder {
+class PlainBlockBuilder final : public BlockBuilder {
public:
explicit PlainBlockBuilder(const WriterOptions *options)
: options_(options) {
@@ -109,7 +109,7 @@ class PlainBlockBuilder : public BlockBuilder {
// A plain decoder for generic fixed size data types.
//
template<DataType Type>
-class PlainBlockDecoder : public BlockDecoder {
+class PlainBlockDecoder final : public BlockDecoder {
public:
explicit PlainBlockDecoder(Slice slice)
: data_(std::move(slice)),
http://git-wip-us.apache.org/repos/asf/kudu/blob/eb4b46b8/src/kudu/cfile/rle_block.h
----------------------------------------------------------------------
diff --git a/src/kudu/cfile/rle_block.h b/src/kudu/cfile/rle_block.h
index b441529..c7702f1 100644
--- a/src/kudu/cfile/rle_block.h
+++ b/src/kudu/cfile/rle_block.h
@@ -46,7 +46,7 @@ enum {
// RLE encoder for the BOOL datatype: uses an RLE-encoded bitmap to
// represent a bool column.
//
-class RleBitMapBlockBuilder : public BlockBuilder {
+class RleBitMapBlockBuilder final : public BlockBuilder {
public:
explicit RleBitMapBlockBuilder(const WriterOptions* options)
: encoder_(&buf_, 1),
@@ -107,7 +107,7 @@ class RleBitMapBlockBuilder : public BlockBuilder {
//
// RLE decoder for bool datatype
//
-class RleBitMapBlockDecoder : public BlockDecoder {
+class RleBitMapBlockDecoder final : public BlockDecoder {
public:
explicit RleBitMapBlockDecoder(Slice slice)
: data_(std::move(slice)),
@@ -213,7 +213,7 @@ class RleBitMapBlockDecoder : public BlockDecoder {
// TODO : consider if this can also be used for BOOL with only minor
// alterations
template <DataType IntType>
-class RleIntBlockBuilder : public BlockBuilder {
+class RleIntBlockBuilder final : public BlockBuilder {
public:
explicit RleIntBlockBuilder(const WriterOptions* options)
: rle_encoder_(&buf_, kCppTypeSize * 8),
@@ -296,7 +296,7 @@ class RleIntBlockBuilder : public BlockBuilder {
// code here for the BOOL type.
//
template <DataType IntType>
-class RleIntBlockDecoder : public BlockDecoder {
+class RleIntBlockDecoder final : public BlockDecoder {
public:
explicit RleIntBlockDecoder(Slice slice)
: data_(std::move(slice)),