You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2018/12/07 03:47:51 UTC
[2/5] kudu git commit: (02/05) delta_store: convert
DeltaIterator::PrepareBatch flags into bitfield
(02/05) delta_store: convert DeltaIterator::PrepareBatch flags into bitfield
A follow-on patch will introduce DeltaIterator::SelectUpdates(), a new
method for extracting delta information from a DeltaIterator. As it is
only to be used by diff scans and will require dedicated in-memory state,
it'll also come with a new DeltaIterator::PrepareBatch() flag.
However, diff scans will need to use both SelectUpdates() and
ApplyUpdates(), and it'd be a shame to have to reseek and reprepare just to
use both extraction methods. As such, this patch makes it possible to
prepare a batch for use in multiple delta extraction methods.
Change-Id: Id500ec3cc9459a78ae1098be3f3cd0cacb7a7a1a
Reviewed-on: http://gerrit.cloudera.org:8080/11857
Tested-by: Kudu Jenkins
Reviewed-by: Mike Percy <mp...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/d927675b
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/d927675b
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/d927675b
Branch: refs/heads/master
Commit: d927675b0af52adfd03a52265337ee1dc1609fa4
Parents: 0008990
Author: Adar Dembo <ad...@cloudera.com>
Authored: Tue Sep 11 16:27:56 2018 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Fri Dec 7 03:43:24 2018 +0000
----------------------------------------------------------------------
src/kudu/tablet/delta_iterator_merger.cc | 4 +-
src/kudu/tablet/delta_iterator_merger.h | 2 +-
src/kudu/tablet/delta_store.cc | 34 ++++++---------
src/kudu/tablet/delta_store.h | 62 ++++++++++++---------------
src/kudu/tablet/deltafile.cc | 4 +-
src/kudu/tablet/deltafile.h | 2 +-
src/kudu/tablet/deltamemstore.cc | 4 +-
src/kudu/tablet/deltamemstore.h | 2 +-
src/kudu/tablet/tablet-test-util.h | 21 +++------
9 files changed, 55 insertions(+), 80 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_iterator_merger.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_iterator_merger.cc b/src/kudu/tablet/delta_iterator_merger.cc
index 23ed670..f6f5840 100644
--- a/src/kudu/tablet/delta_iterator_merger.cc
+++ b/src/kudu/tablet/delta_iterator_merger.cc
@@ -57,9 +57,9 @@ Status DeltaIteratorMerger::SeekToOrdinal(rowid_t idx) {
return Status::OK();
}
-Status DeltaIteratorMerger::PrepareBatch(size_t nrows, PrepareFlag flag) {
+Status DeltaIteratorMerger::PrepareBatch(size_t nrows, int prepare_flags) {
for (const unique_ptr<DeltaIterator> &iter : iters_) {
- RETURN_NOT_OK(iter->PrepareBatch(nrows, flag));
+ RETURN_NOT_OK(iter->PrepareBatch(nrows, prepare_flags));
}
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_iterator_merger.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_iterator_merger.h b/src/kudu/tablet/delta_iterator_merger.h
index e5e3bb1..a8efeb6 100644
--- a/src/kudu/tablet/delta_iterator_merger.h
+++ b/src/kudu/tablet/delta_iterator_merger.h
@@ -60,7 +60,7 @@ class DeltaIteratorMerger : public DeltaIterator {
Status SeekToOrdinal(rowid_t idx) override;
- Status PrepareBatch(size_t nrows, PrepareFlag flag) override;
+ Status PrepareBatch(size_t nrows, int prepare_flags) override;
Status ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) override;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_store.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.cc b/src/kudu/tablet/delta_store.cc
index 13b1e4a..bb7a91f 100644
--- a/src/kudu/tablet/delta_store.cc
+++ b/src/kudu/tablet/delta_store.cc
@@ -107,7 +107,7 @@ DeltaPreparer<Traits>::DeltaPreparer(RowIteratorOptions opts)
: opts_(std::move(opts)),
cur_prepared_idx_(0),
prev_prepared_idx_(0),
- prepared_for_(NOT_PREPARED),
+ prepared_flags_(DeltaIterator::PREPARE_NONE),
deletion_state_(UNKNOWN) {
}
@@ -115,11 +115,13 @@ template<class Traits>
void DeltaPreparer<Traits>::Seek(rowid_t row_idx) {
cur_prepared_idx_ = row_idx;
prev_prepared_idx_ = row_idx;
- prepared_for_ = NOT_PREPARED;
+ prepared_flags_ = DeltaIterator::PREPARE_NONE;
}
template<class Traits>
-void DeltaPreparer<Traits>::Start(DeltaIterator::PrepareFlag flag) {
+void DeltaPreparer<Traits>::Start(int prepare_flags) {
+ DCHECK_NE(prepare_flags, DeltaIterator::PREPARE_NONE);
+ prepared_flags_ = prepare_flags;
if (updates_by_col_.empty()) {
updates_by_col_.resize(opts_.projection->num_columns());
}
@@ -130,16 +132,6 @@ void DeltaPreparer<Traits>::Start(DeltaIterator::PrepareFlag flag) {
reinserted_.clear();
prepared_deltas_.clear();
deletion_state_ = UNKNOWN;
- switch (flag) {
- case DeltaIterator::PREPARE_FOR_APPLY:
- prepared_for_ = PREPARED_FOR_APPLY;
- break;
- case DeltaIterator::PREPARE_FOR_COLLECT:
- prepared_for_ = PREPARED_FOR_COLLECT;
- break;
- default:
- LOG(FATAL) << "Unknown prepare flag: " << flag;
- }
}
template<class Traits>
@@ -157,7 +149,7 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
}
MaybeProcessPreviousRowChange(key.row_idx());
- if (prepared_for_ == PREPARED_FOR_APPLY) {
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY) {
RowChangeListDecoder decoder((RowChangeList(val)));
if (Traits::kInitializeDecodersWithSafetyChecks) {
RETURN_NOT_OK(decoder.Init());
@@ -201,8 +193,8 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
}
}
}
- } else {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_COLLECT);
+ }
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT) {
PreparedDelta d;
d.key = key;
d.val = val;
@@ -216,7 +208,7 @@ Status DeltaPreparer<Traits>::AddDelta(const DeltaKey& key, Slice val, bool* fin
template<class Traits>
Status DeltaPreparer<Traits>::ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, dst->nrows());
const ColumnSchema* col_schema = &opts_.projection->column(col_to_apply);
@@ -236,7 +228,7 @@ Status DeltaPreparer<Traits>::ApplyUpdates(size_t col_to_apply, ColumnBlock* dst
template<class Traits>
Status DeltaPreparer<Traits>::ApplyDeletes(SelectionVector* sel_vec) {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, sel_vec->nrows());
for (const auto& row_id : deleted_) {
@@ -254,7 +246,7 @@ Status DeltaPreparer<Traits>::ApplyDeletes(SelectionVector* sel_vec) {
template<class Traits>
Status DeltaPreparer<Traits>::CollectMutations(vector<Mutation*>* dst, Arena* arena) {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_COLLECT);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_COLLECT);
DCHECK_LE(cur_prepared_idx_ - prev_prepared_idx_, dst->size());
for (const PreparedDelta& src : prepared_deltas_) {
DeltaKey key = src.key;
@@ -306,7 +298,7 @@ Status DeltaPreparer<Traits>::FilterColumnIdsAndCollectDeltas(
template<class Traits>
bool DeltaPreparer<Traits>::MayHaveDeltas() const {
- DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY);
+ DCHECK(prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY);
if (!deleted_.empty()) {
return true;
}
@@ -323,7 +315,7 @@ bool DeltaPreparer<Traits>::MayHaveDeltas() const {
template<class Traits>
void DeltaPreparer<Traits>::MaybeProcessPreviousRowChange(boost::optional<rowid_t> cur_row_idx) {
- if (prepared_for_ == PREPARED_FOR_APPLY &&
+ if (prepared_flags_ & DeltaIterator::PREPARE_FOR_APPLY &&
last_added_idx_ &&
(!cur_row_idx || cur_row_idx != *last_added_idx_)) {
switch (deletion_state_) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/delta_store.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_store.h b/src/kudu/tablet/delta_store.h
index 12ec9c6..955f4a7 100644
--- a/src/kudu/tablet/delta_store.h
+++ b/src/kudu/tablet/delta_store.h
@@ -198,22 +198,37 @@ class DeltaIterator : public PreparedDeltas {
// block, and must be called at least once prior to PrepareBatch().
virtual Status SeekToOrdinal(rowid_t idx) = 0;
- // Argument to PrepareBatch(). See below.
- enum PrepareFlag {
- PREPARE_FOR_APPLY,
- PREPARE_FOR_COLLECT
- };
-
// Prepare to apply deltas to a block of rows. This takes a consistent snapshot
// of all updates to the next 'nrows' rows, so that subsequent calls to a
// PreparedDeltas method will not cause any "tearing"/non-atomicity.
//
- // 'flag' denotes whether the batch will be used for collecting mutations or
- // for applying them. Some implementations may choose to prepare differently.
+ // 'prepare_flags' is a bitfield describing what operation(s) the batch will
+ // be used for; some implementations may choose to prepare differently.
+ // PREPARE_NONE is an invalid value; it is only used internally.
//
// Each time this is called, the iterator is advanced by the full length
// of the previously prepared block.
- virtual Status PrepareBatch(size_t nrows, PrepareFlag flag) = 0;
+ enum {
+ // There are no prepared blocks. Attempts to call a PreparedDeltas function
+ // will fail.
+ PREPARE_NONE = 0,
+
+ // Prepare a batch of deltas for applying. All deltas in the batch will be
+ // decoded. Operations affecting row data (i.e. UPDATEs and REINSERTs) will
+ // be coalesced into a column-major data structure suitable for
+ // ApplyUpdates. Operations affecting row lifecycle (i.e. DELETES and
+ // REINSERTs) will be coalesced into a row-major data structure suitable for ApplyDeletes.
+ //
+ // On success, ApplyUpdates and ApplyDeltas will be callable.
+ PREPARE_FOR_APPLY = 1 << 0,
+
+ // Prepare a batch of deltas for collecting. Deltas will remain encoded and
+ // in the order that they were loaded from the backing store.
+ //
+ // On success, CollectMutations and FilterColumnIdsAndCollectDeltas will be callable.
+ PREPARE_FOR_COLLECT = 1 << 1
+ };
+ virtual Status PrepareBatch(size_t nrows, int prepare_flags) = 0;
// Returns true if there are any more rows left in this iterator.
virtual bool HasNext() = 0;
@@ -266,7 +281,7 @@ class DeltaPreparer : public PreparedDeltas {
// on the part of a DeltaIterator.
//
// Call at the beginning of DeltaIterator::PrepareBatch.
- void Start(DeltaIterator::PrepareFlag flag);
+ void Start(int prepare_flags);
// Updates internal state to reflect the end of delta batch preparation on the
// part of a DeltaIterator.
@@ -329,30 +344,9 @@ class DeltaPreparer : public PreparedDeltas {
boost::optional<rowid_t> last_added_idx_;
// Whether there are any prepared blocks.
- enum PreparedFor {
- // There are no prepared blocks. Attempts to call a PreparedDeltas function
- // will fail.
- NOT_PREPARED,
-
- // The DeltaPreparer has prepared a batch of deltas for applying. All deltas
- // in the batch have been decoded. Operations affecting row data (i.e.
- // UPDATEs and REINSERTs) have been coalesced into a column-major data
- // structure suitable for ApplyUpdates. Operations affecting row lifecycle
- // (i.e. DELETES and REINSERTs) have been coalesced into a row-major data
- // structure suitable for ApplyDeletes.
- //
- // ApplyUpdates and ApplyDeltas are now callable.
- PREPARED_FOR_APPLY,
-
- // The DeltaPreparer has prepared a batch of deltas for collecting. Deltas
- // remain encoded and in the order that they were loaded from the backing store.
- //
- // CollectMutations and FilterColumnIdsAndCollectDeltas are now callable.
- PREPARED_FOR_COLLECT
- };
- PreparedFor prepared_for_;
+ int prepared_flags_;
- // State when prepared_for_ == PREPARED_FOR_APPLY
+ // State when prepared_flags_ & PREPARED_FOR_APPLY
// ------------------------------------------------------------
struct ColumnUpdate {
rowid_t row_id;
@@ -372,7 +366,7 @@ class DeltaPreparer : public PreparedDeltas {
};
RowDeletionState deletion_state_;
- // State when prepared_for_ == PREPARED_FOR_COLLECT
+ // State when prepared_flags_ & PREPARED_FOR_COLLECT
// ------------------------------------------------------------
struct PreparedDelta {
DeltaKey key;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltafile.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.cc b/src/kudu/tablet/deltafile.cc
index 575895a..7ca6719 100644
--- a/src/kudu/tablet/deltafile.cc
+++ b/src/kudu/tablet/deltafile.cc
@@ -532,7 +532,7 @@ string DeltaFileIterator<Type>::PreparedDeltaBlock::ToString() const {
}
template<DeltaType Type>
-Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, PrepareFlag flag) {
+Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, int prepare_flags) {
DCHECK(initted_) << "Must call Init()";
DCHECK(exhausted_ || index_iter_) << "Must call SeekToOrdinal()";
@@ -587,7 +587,7 @@ Status DeltaFileIterator<Type>::PrepareBatch(size_t nrows, PrepareFlag flag) {
#endif
prepared_ = true;
- preparer_.Start(flag);
+ preparer_.Start(prepare_flags);
RETURN_NOT_OK(AddDeltas(start_row, stop_row));
preparer_.Finish(nrows);
return Status::OK();
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltafile.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltafile.h b/src/kudu/tablet/deltafile.h
index e99a4f4..217dd4c 100644
--- a/src/kudu/tablet/deltafile.h
+++ b/src/kudu/tablet/deltafile.h
@@ -227,7 +227,7 @@ class DeltaFileIterator : public DeltaIterator {
Status SeekToOrdinal(rowid_t idx) override;
- Status PrepareBatch(size_t nrows, PrepareFlag flag) override;
+ Status PrepareBatch(size_t nrows, int prepare_flags) override;
Status ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) override;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltamemstore.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltamemstore.cc b/src/kudu/tablet/deltamemstore.cc
index 8676d33..a25c110 100644
--- a/src/kudu/tablet/deltamemstore.cc
+++ b/src/kudu/tablet/deltamemstore.cc
@@ -217,7 +217,7 @@ Status DMSIterator::SeekToOrdinal(rowid_t row_idx) {
return Status::OK();
}
-Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) {
+Status DMSIterator::PrepareBatch(size_t nrows, int prepare_flags) {
// This current implementation copies the whole batch worth of deltas
// into a buffer local to this iterator, after filtering out deltas which
// aren't yet committed in the current MVCC snapshot. The theory behind
@@ -233,8 +233,8 @@ Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) {
DCHECK(initted_) << "must init";
rowid_t start_row = preparer_.cur_prepared_idx();
rowid_t stop_row = start_row + nrows - 1;
- preparer_.Start(flag);
+ preparer_.Start(prepare_flags);
bool finished_row = false;
while (iter_->IsValid()) {
Slice key_slice, val;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/deltamemstore.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/deltamemstore.h b/src/kudu/tablet/deltamemstore.h
index d9f094b..a5a92f7 100644
--- a/src/kudu/tablet/deltamemstore.h
+++ b/src/kudu/tablet/deltamemstore.h
@@ -199,7 +199,7 @@ class DMSIterator : public DeltaIterator {
Status SeekToOrdinal(rowid_t row_idx) override;
- Status PrepareBatch(size_t nrows, PrepareFlag flag) override;
+ Status PrepareBatch(size_t nrows, int prepare_flags) override;
Status ApplyUpdates(size_t col_to_apply, ColumnBlock* dst,
const SelectionVector& filter) override;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d927675b/src/kudu/tablet/tablet-test-util.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test-util.h b/src/kudu/tablet/tablet-test-util.h
index 89a816e..87daee6 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -809,15 +809,15 @@ void RunDeltaFuzzTest(const DeltaStore& store,
std::unique_ptr<DeltaIterator> iter(raw_iter);
ASSERT_OK(iter->Init(nullptr));
- // Run PREPARE_FOR_APPLY method tests in batches, in case there's some bug
- // related to batching.
+ // Run tests in batches, in case there's some bug related to batching.
ASSERT_OK(iter->SeekToOrdinal(0));
rowid_t start_row_idx = 0;
while (iter->HasNext()) {
int batch_size = prng->Uniform(kMaxBatchSize) + 1;
- SCOPED_TRACE(strings::Substitute("APPLY: batch starting at $0 ($1 rows)",
+ SCOPED_TRACE(strings::Substitute("batch starting at $0 ($1 rows)",
start_row_idx, batch_size));
- ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_APPLY));
+ ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_APPLY |
+ DeltaIterator::PREPARE_FOR_COLLECT));
// Test ApplyDeletes: the selection vector is all true and a row is unset
// if the last relevant update deleted it.
@@ -851,18 +851,6 @@ void RunDeltaFuzzTest(const DeltaStore& store,
<< "\nActual column block: " << actual_scb.ToString();
}
- start_row_idx += batch_size;
- }
-
- // Now run PREPARE_FOR_COLLECT method tests.
- ASSERT_OK(iter->SeekToOrdinal(0));
- start_row_idx = 0;
- while (iter->HasNext()) {
- int batch_size = prng->Uniform(kMaxBatchSize - 1) + 1;
- SCOPED_TRACE(strings::Substitute("COLLECT: batch starting at $0 ($1 rows)",
- start_row_idx, batch_size));
- ASSERT_OK(iter->PrepareBatch(batch_size, DeltaIterator::PREPARE_FOR_COLLECT));
-
// Test CollectMutations: all relevant updates are returned.
{
Arena arena(1024);
@@ -917,6 +905,7 @@ void RunDeltaFuzzTest(const DeltaStore& store,
ASSERT_EQ(expected_deltas[j].second, actual_deltas[j].cell);
}
}
+
start_row_idx += batch_size;
}
}