Repository: kudu Updated Branches: refs/heads/master 521189744 -> 0fbebd67c
tablet: encapsulate common iterator options I intend to introduce an additional option or two as part of the "diff scan" API. To that end, this patch knocks the plumbing out of the way by moving all of the common iterator options into a new struct. Unfortunately, I can't use it in Tablet::NewRowIterator because while nearly every iterator expects a pointer to the projection, Tablet::Iterator stores a copy of the projection itself. Change-Id: I7232d163436e69999bba75ed66756d2a86c5a959 Reviewed-on: http://gerrit.cloudera.org:8080/10802 Tested-by: Adar Dembo <[email protected]> Reviewed-by: Mike Percy <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kudu/repo Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/0fbebd67 Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/0fbebd67 Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/0fbebd67 Branch: refs/heads/master Commit: 0fbebd67c2ec5dd93775a3c3454a1e08496e5179 Parents: 5211897 Author: Adar Dembo <[email protected]> Authored: Fri Jun 22 12:13:28 2018 -0700 Committer: Adar Dembo <[email protected]> Committed: Thu Jul 19 18:53:13 2018 +0000 ---------------------------------------------------------------------- src/kudu/tablet/compaction.cc | 18 ++++-- src/kudu/tablet/delta_compaction-test.cc | 10 ++-- src/kudu/tablet/delta_iterator_merger.cc | 7 +-- src/kudu/tablet/delta_iterator_merger.h | 10 ++-- src/kudu/tablet/delta_store.h | 10 ++-- src/kudu/tablet/delta_tracker.cc | 25 ++++----- src/kudu/tablet/delta_tracker.h | 24 ++++---- src/kudu/tablet/deltafile-test.cc | 31 ++++++---- src/kudu/tablet/deltafile.cc | 59 ++++++++++---------- src/kudu/tablet/deltafile.h | 21 +++---- src/kudu/tablet/deltamemstore-test.cc | 25 ++++++--- src/kudu/tablet/deltamemstore.cc | 28 +++++----- src/kudu/tablet/deltamemstore.h | 23 +++----- src/kudu/tablet/diskrowset-test-base.h | 19 ++++--- src/kudu/tablet/diskrowset-test.cc | 14 +++-- src/kudu/tablet/diskrowset.cc | 19 +++---- src/kudu/tablet/diskrowset.h | 5 +- src/kudu/tablet/memrowset-test.cc | 10 +++- src/kudu/tablet/memrowset.cc | 35 ++++++------ src/kudu/tablet/memrowset.h | 25 ++++----- src/kudu/tablet/mock-rowsets.h | 4 +- .../tablet/mt-rowset_delta_compaction-test.cc | 9 ++- src/kudu/tablet/rowset.cc | 22 +++++--- src/kudu/tablet/rowset.h | 41 ++++++++++---- src/kudu/tablet/tablet-test-util.h | 7 ++- src/kudu/tablet/tablet.cc | 11 +++- src/kudu/tablet/tablet.h | 6 +- 27 files changed, 272 insertions(+), 246 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/compaction.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc index afabfd3..ec9edb4 100644 --- a/src/kudu/tablet/compaction.cc +++ b/src/kudu/tablet/compaction.cc @@ -90,9 +90,12 @@ class MemRowSetCompactionInput : public CompactionInput { MemRowSetCompactionInput(const MemRowSet& memrowset, const MvccSnapshot& snap, const Schema* projection) - : iter_(memrowset.NewIterator(projection, snap)), - arena_(32*1024), + : arena_(32*1024), has_more_blocks_(false) { + RowIteratorOptions opts; + opts.projection = projection; + opts.snap = snap; + iter_.reset(memrowset.NewIterator(opts)); } Status Init() override { @@ -856,15 +859,20 @@ Status CompactionInput::Create(const DiskRowSet &rowset, gscoped_ptr<RowwiseIterator> base_iter(new MaterializingIterator(base_cwise)); // Creates a DeltaIteratorMerger that will only include the relevant REDO deltas. + RowIteratorOptions redo_opts; + redo_opts.projection = projection; + redo_opts.snap = snap; unique_ptr<DeltaIterator> redo_deltas; RETURN_NOT_OK_PREPEND(rowset.delta_tracker_->NewDeltaIterator( - projection, snap, DeltaTracker::REDOS_ONLY, &redo_deltas), "Could not open REDOs"); + redo_opts, DeltaTracker::REDOS_ONLY, &redo_deltas), "Could not open REDOs"); // Creates a DeltaIteratorMerger that will only include UNDO deltas. Using the // "empty" snapshot ensures that all deltas are included. + RowIteratorOptions undo_opts; + undo_opts.projection = projection; + undo_opts.snap = MvccSnapshot::CreateSnapshotIncludingNoTransactions(); unique_ptr<DeltaIterator> undo_deltas; RETURN_NOT_OK_PREPEND(rowset.delta_tracker_->NewDeltaIterator( - projection, MvccSnapshot::CreateSnapshotIncludingNoTransactions(), - DeltaTracker::UNDOS_ONLY, &undo_deltas), "Could not open UNDOs"); + undo_opts, DeltaTracker::UNDOS_ONLY, &undo_deltas), "Could not open UNDOs"); out->reset(new DiskRowSetCompactionInput(std::move(base_iter), std::move(redo_deltas), http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/delta_compaction-test.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/delta_compaction-test.cc b/src/kudu/tablet/delta_compaction-test.cc index 2c22955..21be212 100644 --- a/src/kudu/tablet/delta_compaction-test.cc +++ b/src/kudu/tablet/delta_compaction-test.cc @@ -42,7 +42,7 @@ #include "kudu/tablet/delta_stats.h" #include "kudu/tablet/delta_store.h" #include "kudu/tablet/deltafile.h" -#include "kudu/tablet/mvcc.h" +#include "kudu/tablet/rowset.h" #include "kudu/util/faststring.h" #include "kudu/util/slice.h" #include "kudu/util/status.h" @@ -194,11 +194,11 @@ TEST_F(TestDeltaCompaction, TestMergeMultipleSchemas) { } // Merge - MvccSnapshot snap(MvccSnapshot::CreateSnapshotIncludingAllTransactions()); const Schema& merge_schema = schemas.back(); + RowIteratorOptions opts; + opts.projection = &merge_schema; unique_ptr<DeltaIterator> merge_iter; - ASSERT_OK(DeltaIteratorMerger::Create(inputs, &merge_schema, - snap, &merge_iter)); + ASSERT_OK(DeltaIteratorMerger::Create(inputs, opts, &merge_iter)); gscoped_ptr<DeltaFileWriter> dfw; BlockId block_id; ASSERT_OK(GetDeltaFileWriter(&dfw, &block_id)); @@ -210,7 +210,7 @@ TEST_F(TestDeltaCompaction, TestMergeMultipleSchemas) { shared_ptr<DeltaFileReader> dfr; ASSERT_OK(GetDeltaFileReader(block_id, &dfr)); DeltaIterator* raw_iter; - ASSERT_OK(dfr->NewDeltaIterator(&merge_schema, snap, &raw_iter)); + ASSERT_OK(dfr->NewDeltaIterator(opts, &raw_iter)); gscoped_ptr<DeltaIterator> scoped_iter(raw_iter); vector<string> results; http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/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 b287fda..23e6fff 100644 --- a/src/kudu/tablet/delta_iterator_merger.cc +++ b/src/kudu/tablet/delta_iterator_merger.cc @@ -26,14 +26,12 @@ namespace kudu { class Arena; class ColumnBlock; -class Schema; class SelectionVector; struct ColumnId; namespace tablet { class Mutation; -class MvccSnapshot; using std::shared_ptr; using std::string; @@ -148,14 +146,13 @@ string DeltaIteratorMerger::ToString() const { Status DeltaIteratorMerger::Create( const vector<shared_ptr<DeltaStore> > &stores, - const Schema* projection, - const MvccSnapshot &snapshot, + const RowIteratorOptions& opts, unique_ptr<DeltaIterator>* out) { vector<unique_ptr<DeltaIterator> > delta_iters; for (const shared_ptr<DeltaStore> &store : stores) { DeltaIterator* raw_iter; - Status s = store->NewDeltaIterator(projection, snapshot, &raw_iter); + Status s = store->NewDeltaIterator(opts, &raw_iter); if (s.IsNotFound()) { continue; } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/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 2d77400..d1bd9b4 100644 --- a/src/kudu/tablet/delta_iterator_merger.h +++ b/src/kudu/tablet/delta_iterator_merger.h @@ -18,8 +18,8 @@ #define KUDU_TABLET_DELTA_ITERATOR_MERGER_H #include <cstddef> -#include <string> #include <memory> +#include <string> #include <vector> #include "kudu/common/rowid.h" @@ -32,14 +32,13 @@ namespace kudu { class Arena; class ColumnBlock; class ScanSpec; -class Schema; class SelectionVector; struct ColumnId; namespace tablet { class Mutation; -class MvccSnapshot; +struct RowIteratorOptions; // DeltaIterator that simply combines together other DeltaIterators, // applying deltas from each in order. @@ -51,9 +50,8 @@ class DeltaIteratorMerger : public DeltaIterator { // If only one store is input, this will automatically return an unwrapped // iterator for greater efficiency. static Status Create( - const std::vector<std::shared_ptr<DeltaStore> > &stores, - const Schema* projection, - const MvccSnapshot &snapshot, + const std::vector<std::shared_ptr<DeltaStore>> &stores, + const RowIteratorOptions& opts, std::unique_ptr<DeltaIterator>* out); //////////////////////////////////////////////////////////// http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/delta_store.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/delta_store.h b/src/kudu/tablet/delta_store.h index 9f5416b..aea8a39 100644 --- a/src/kudu/tablet/delta_store.h +++ b/src/kudu/tablet/delta_store.h @@ -44,6 +44,7 @@ class DeltaIterator; class DeltaStats; class Mutation; class MvccSnapshot; +struct RowIteratorOptions; // Interface for the pieces of the system that track deltas/updates. // This is implemented by DeltaMemStore and by DeltaFileReader. @@ -58,17 +59,16 @@ class DeltaStore { // Create a DeltaIterator for the given projection. // - // The projection corresponds to whatever scan is currently ongoing. + // The projection in 'opts' corresponds to whatever scan is currently ongoing. // All RowBlocks passed to this DeltaIterator must have this same schema. // - // 'snapshot' is the MVCC state which determines which transactions + // The snapshot in 'opts' is the MVCC state which determines which transactions // should be considered committed (and thus applied by the iterator). // // Returns Status::OK and sets 'iterator' to the new DeltaIterator, or // returns Status::NotFound if the mutations within this delta store - // cannot include 'snap'. - virtual Status NewDeltaIterator(const Schema *projection, - const MvccSnapshot &snap, + // cannot include the snapshot. + virtual Status NewDeltaIterator(const RowIteratorOptions& opts, DeltaIterator** iterator) const = 0; // Set *deleted to true if the latest update for the given row is a deletion. http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/delta_tracker.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/delta_tracker.cc b/src/kudu/tablet/delta_tracker.cc index 96d581e..0ab956f 100644 --- a/src/kudu/tablet/delta_tracker.cc +++ b/src/kudu/tablet/delta_tracker.cc @@ -46,7 +46,6 @@ #include "kudu/tablet/delta_store.h" #include "kudu/tablet/deltafile.h" #include "kudu/tablet/deltamemstore.h" -#include "kudu/tablet/mvcc.h" #include "kudu/tablet/rowset.h" #include "kudu/tablet/rowset_metadata.h" #include "kudu/tablet/tablet.pb.h" @@ -180,9 +179,9 @@ Status DeltaTracker::MakeDeltaIteratorMergerUnlocked(size_t start_idx, size_t en target_stores->push_back(delta_store); target_blocks->push_back(dfr->block_id()); } - RETURN_NOT_OK(DeltaIteratorMerger::Create( - inputs, projection, - MvccSnapshot::CreateSnapshotIncludingAllTransactions(), out)); + RowIteratorOptions opts; + opts.projection = projection; + RETURN_NOT_OK(DeltaIteratorMerger::Create(inputs, opts, out)); return Status::OK(); } @@ -549,20 +548,18 @@ void DeltaTracker::CollectStores(vector<shared_ptr<DeltaStore>>* deltas, } } -Status DeltaTracker::NewDeltaIterator(const Schema* schema, - const MvccSnapshot& snap, +Status DeltaTracker::NewDeltaIterator(const RowIteratorOptions& opts, WhichStores which, unique_ptr<DeltaIterator>* out) const { - std::vector<shared_ptr<DeltaStore> > stores; + std::vector<shared_ptr<DeltaStore>> stores; CollectStores(&stores, which); - return DeltaIteratorMerger::Create(stores, schema, snap, out); + return DeltaIteratorMerger::Create(stores, opts, out); } Status DeltaTracker::NewDeltaFileIterator( - const Schema* schema, - const MvccSnapshot& snap, + const RowIteratorOptions& opts, DeltaType type, - vector<shared_ptr<DeltaStore> >* included_stores, + vector<shared_ptr<DeltaStore>>* included_stores, unique_ptr<DeltaIterator>* out) const { { std::lock_guard<rw_spinlock> lock(component_lock_); @@ -585,14 +582,14 @@ Status DeltaTracker::NewDeltaFileIterator( ignore_result(down_cast<DeltaFileReader*>(store.get())); } - return DeltaIteratorMerger::Create(*included_stores, schema, snap, out); + return DeltaIteratorMerger::Create(*included_stores, opts, out); } Status DeltaTracker::WrapIterator(const shared_ptr<CFileSet::Iterator> &base, - const MvccSnapshot &mvcc_snap, + const RowIteratorOptions& opts, gscoped_ptr<ColumnwiseIterator>* out) const { unique_ptr<DeltaIterator> iter; - RETURN_NOT_OK(NewDeltaIterator(&base->schema(), mvcc_snap, &iter)); + RETURN_NOT_OK(NewDeltaIterator(opts, &iter)); out->reset(new DeltaApplier(base, std::move(iter))); return Status::OK(); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/delta_tracker.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/delta_tracker.h b/src/kudu/tablet/delta_tracker.h index 4732c38..ed32c4b 100644 --- a/src/kudu/tablet/delta_tracker.h +++ b/src/kudu/tablet/delta_tracker.h @@ -63,11 +63,11 @@ namespace tablet { class DeltaFileReader; class DeltaMemStore; -class MvccSnapshot; class OperationResultPB; class RowSetMetadata; class RowSetMetadataUpdate; struct ProbeStats; +struct RowIteratorOptions; // The DeltaTracker is the part of a DiskRowSet which is responsible for // tracking modifications against the base data. It consists of a set of @@ -88,7 +88,7 @@ class DeltaTracker { gscoped_ptr<DeltaTracker>* delta_tracker); Status WrapIterator(const std::shared_ptr<CFileSet::Iterator> &base, - const MvccSnapshot &mvcc_snap, + const RowIteratorOptions& opts, gscoped_ptr<ColumnwiseIterator>* out) const; // Enum used for NewDeltaIterator() and CollectStores() below. @@ -104,17 +104,14 @@ class DeltaTracker { // by this DeltaTracker. Depending on the value of 'which' (see above), // this iterator may include UNDOs, REDOs, or both. // - // 'schema' is the schema of the rows that are being read by the client. - // It must remain valid for the lifetime of the returned iterator. - Status NewDeltaIterator(const Schema* schema, - const MvccSnapshot& snap, + // Pointers in 'opts' must remain valid for the lifetime of the returned iterator. + Status NewDeltaIterator(const RowIteratorOptions& opts, WhichStores which, std::unique_ptr<DeltaIterator>* out) const; - Status NewDeltaIterator(const Schema* schema, - const MvccSnapshot& snap, + Status NewDeltaIterator(const RowIteratorOptions& opts, std::unique_ptr<DeltaIterator>* out) const { - return NewDeltaIterator(schema, snap, UNDOS_AND_REDOS, out); + return NewDeltaIterator(opts, UNDOS_AND_REDOS, out); } @@ -122,11 +119,10 @@ class DeltaTracker { // the DMS. // Returns the delta stores being merged in *included_stores. Status NewDeltaFileIterator( - const Schema* schema, - const MvccSnapshot &snap, - DeltaType type, - std::vector<std::shared_ptr<DeltaStore> >* included_stores, - std::unique_ptr<DeltaIterator>* out) const; + const RowIteratorOptions& opts, + DeltaType type, + std::vector<std::shared_ptr<DeltaStore>>* included_stores, + std::unique_ptr<DeltaIterator>* out) const; // Flushes the current DeltaMemStore and replaces it with a new one. // Caller selects whether to also have the RowSetMetadata (and consequently http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/deltafile-test.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/deltafile-test.cc b/src/kudu/tablet/deltafile-test.cc index 0d223d4..3dbc9e6 100644 --- a/src/kudu/tablet/deltafile-test.cc +++ b/src/kudu/tablet/deltafile-test.cc @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +#include "kudu/tablet/deltafile.h" + #include <algorithm> #include <cstddef> #include <cstdint> @@ -45,9 +47,9 @@ #include "kudu/tablet/delta_key.h" #include "kudu/tablet/delta_stats.h" #include "kudu/tablet/delta_store.h" -#include "kudu/tablet/deltafile.h" #include "kudu/tablet/mutation.h" #include "kudu/tablet/mvcc.h" +#include "kudu/tablet/rowset.h" #include "kudu/util/faststring.h" #include "kudu/util/memory/arena.h" #include "kudu/util/status.h" @@ -147,11 +149,13 @@ class TestDeltaFile : public KuduTest { Status OpenDeltaFileIteratorFromReader(DeltaType type, const shared_ptr<DeltaFileReader>& reader, gscoped_ptr<DeltaIterator>* out) { - MvccSnapshot snap = type == REDO ? - MvccSnapshot::CreateSnapshotIncludingAllTransactions() : - MvccSnapshot::CreateSnapshotIncludingNoTransactions(); + RowIteratorOptions opts; + opts.snap = type == REDO ? + MvccSnapshot::CreateSnapshotIncludingAllTransactions() : + MvccSnapshot::CreateSnapshotIncludingNoTransactions(); + opts.projection = &schema_; DeltaIterator* raw_iter; - RETURN_NOT_OK(reader->NewDeltaIterator(&schema_, snap, &raw_iter)); + RETURN_NOT_OK(reader->NewDeltaIterator(opts, &raw_iter)); out->reset(raw_iter); return Status::OK(); } @@ -333,25 +337,28 @@ TEST_F(TestDeltaFile, TestSkipsDeltasOutOfRange) { gscoped_ptr<DeltaIterator> iter; + RowIteratorOptions opts; + opts.projection = &schema_; + // should skip - MvccSnapshot snap1(Timestamp(9)); - ASSERT_FALSE(snap1.MayHaveCommittedTransactionsAtOrAfter(Timestamp(10))); + opts.snap = MvccSnapshot(Timestamp(9)); + ASSERT_FALSE(opts.snap.MayHaveCommittedTransactionsAtOrAfter(Timestamp(10))); DeltaIterator* raw_iter = nullptr; - Status s = reader->NewDeltaIterator(&schema_, snap1, &raw_iter); + Status s = reader->NewDeltaIterator(opts, &raw_iter); ASSERT_TRUE(s.IsNotFound()); ASSERT_TRUE(raw_iter == nullptr); // should include raw_iter = nullptr; - MvccSnapshot snap2(Timestamp(15)); - ASSERT_OK(reader->NewDeltaIterator(&schema_, snap2, &raw_iter)); + opts.snap = MvccSnapshot(Timestamp(15)); + ASSERT_OK(reader->NewDeltaIterator(opts, &raw_iter)); ASSERT_TRUE(raw_iter != nullptr); iter.reset(raw_iter); // should include raw_iter = nullptr; - MvccSnapshot snap3(Timestamp(21)); - ASSERT_OK(reader->NewDeltaIterator(&schema_, snap3, &raw_iter)); + opts.snap = MvccSnapshot(Timestamp(21)); + ASSERT_OK(reader->NewDeltaIterator(opts, &raw_iter)); ASSERT_TRUE(raw_iter != nullptr); iter.reset(raw_iter); } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/deltafile.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/deltafile.cc b/src/kudu/tablet/deltafile.cc index fe504ba..d00c654 100644 --- a/src/kudu/tablet/deltafile.cc +++ b/src/kudu/tablet/deltafile.cc @@ -315,25 +315,24 @@ Status DeltaFileReader::CloneForDebugging(FsManager* fs_manager, return DeltaFileReader::OpenNoInit(std::move(block), delta_type_, options, out); } -Status DeltaFileReader::NewDeltaIterator(const Schema *projection, - const MvccSnapshot &snap, +Status DeltaFileReader::NewDeltaIterator(const RowIteratorOptions& opts, DeltaIterator** iterator) const { - if (IsRelevantForSnapshot(snap)) { + if (IsRelevantForSnapshot(opts.snap)) { if (VLOG_IS_ON(2)) { if (!init_once_.init_succeeded()) { TRACE_COUNTER_INCREMENT("delta_iterators_lazy_initted", 1); VLOG(2) << (delta_type_ == REDO ? "REDO" : "UNDO") << " delta " << ToString() << " has no delta stats" - << ": can't cull for " << snap.ToString(); + << ": can't cull for " << opts.snap.ToString(); } else if (delta_type_ == REDO) { VLOG(2) << "REDO delta " << ToString() << " has min ts " << delta_stats_->min_timestamp().ToString() - << ": can't cull for " << snap.ToString(); + << ": can't cull for " << opts.snap.ToString(); } else { VLOG(2) << "UNDO delta " << ToString() << " has max ts " << delta_stats_->max_timestamp().ToString() - << ": can't cull for " << snap.ToString(); + << ": can't cull for " << opts.snap.ToString(); } } @@ -341,14 +340,13 @@ Status DeltaFileReader::NewDeltaIterator(const Schema *projection, // Ugly cast, but it lets the iterator fully initialize the reader // during its first seek. *iterator = new DeltaFileIterator( - const_cast<DeltaFileReader*>(this)->shared_from_this(), projection, snap, delta_type_); + const_cast<DeltaFileReader*>(this)->shared_from_this(), opts, delta_type_); return Status::OK(); - } else { - VLOG(2) << "Culling " - << ((delta_type_ == REDO) ? "REDO":"UNDO") - << " delta " << ToString() << " for " << snap.ToString(); - return Status::NotFound("MvccSnapshot outside the range of this delta."); } + VLOG(2) << "Culling " + << ((delta_type_ == REDO) ? "REDO":"UNDO") + << " delta " << ToString() << " for " << opts.snap.ToString(); + return Status::NotFound("MvccSnapshot outside the range of this delta."); } Status DeltaFileReader::CheckRowDeleted(rowid_t row_idx, bool *deleted) const { @@ -361,14 +359,14 @@ Status DeltaFileReader::CheckRowDeleted(rowid_t row_idx, bool *deleted) const { return Status::OK(); } - MvccSnapshot snap_all(MvccSnapshot::CreateSnapshotIncludingAllTransactions()); - - // TODO: would be nice to avoid allocation here, but we don't want to + // TODO(todd): would be nice to avoid allocation here, but we don't want to // duplicate all the logic from NewDeltaIterator. So, we'll heap-allocate // for now. Schema empty_schema; + RowIteratorOptions opts; + opts.projection = &empty_schema; DeltaIterator* raw_iter; - Status s = NewDeltaIterator(&empty_schema, snap_all, &raw_iter); + Status s = NewDeltaIterator(opts, &raw_iter); if (s.IsNotFound()) { *deleted = false; return Status::OK(); @@ -401,11 +399,10 @@ uint64_t DeltaFileReader::EstimateSize() const { //////////////////////////////////////////////////////////// DeltaFileIterator::DeltaFileIterator(shared_ptr<DeltaFileReader> dfr, - const Schema *projection, - MvccSnapshot snap, DeltaType delta_type) + RowIteratorOptions opts, + DeltaType delta_type) : dfr_(std::move(dfr)), - projection_(projection), - mvcc_snap_(std::move(snap)), + opts_(std::move(opts)), prepared_idx_(0xdeadbeef), prepared_count_(0), prepared_(false), @@ -436,7 +433,7 @@ Status DeltaFileIterator::SeekToOrdinal(rowid_t idx) { // that we are querying. We did this already before creating the // DeltaFileIterator, but due to lazy initialization, it's possible // that we weren't able to check at that time. - if (!dfr_->IsRelevantForSnapshot(mvcc_snap_)) { + if (!dfr_->IsRelevantForSnapshot(opts_.snap)) { exhausted_ = true; delta_blocks_.clear(); return Status::OK(); @@ -642,7 +639,7 @@ Status DeltaFileIterator::VisitMutations(Visitor *visitor) { RowChangeList rcl(slice); DVLOG(3) << "Visited " << DeltaType_Name(delta_type_) << " delta for key: " << key.ToString() << " Mut: " - << rcl.ToString(*projection_) << " Continue?: " + << rcl.ToString(*opts_.projection) << " Continue?: " << (continue_visit ? "TRUE" : "FALSE"); } } @@ -692,10 +689,10 @@ struct ApplyingVisitor { int64_t rel_idx = key.row_idx() - dfi->prepared_idx_; DCHECK_GE(rel_idx, 0); - // TODO: this code looks eerily similar to DMSIterator::ApplyUpdates! + // TODO(todd): this code looks eerily similar to DMSIterator::ApplyUpdates! // I bet it can be combined. - const Schema* schema = dfi->projection_; + const Schema* schema = dfi->opts_.projection; RowChangeListDecoder decoder((RowChangeList(deltas))); RETURN_NOT_OK(decoder.Init()); if (decoder.is_update() || decoder.is_reinsert()) { @@ -716,7 +713,7 @@ template<> inline Status ApplyingVisitor<REDO>::Visit(const DeltaKey& key, const Slice& deltas, bool* continue_visit) { - if (IsRedoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) { + if (IsRedoRelevant(dfi->opts_.snap, key.timestamp(), continue_visit)) { DVLOG(3) << "Applied redo delta"; return ApplyMutation(key, deltas); } @@ -728,7 +725,7 @@ template<> inline Status ApplyingVisitor<UNDO>::Visit(const DeltaKey& key, const Slice& deltas, bool* continue_visit) { - if (IsUndoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) { + if (IsUndoRelevant(dfi->opts_.snap, key.timestamp(), continue_visit)) { DVLOG(3) << "Applied undo delta"; return ApplyMutation(key, deltas); } @@ -790,7 +787,7 @@ template<> inline Status LivenessVisitor<REDO>::Visit(const DeltaKey& key, const Slice& deltas, bool* continue_visit) { - if (IsRedoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) { + if (IsRedoRelevant(dfi->opts_.snap, key.timestamp(), continue_visit)) { return ApplyDelete(key, deltas); } return Status::OK(); @@ -800,7 +797,7 @@ template<> inline Status LivenessVisitor<UNDO>::Visit(const DeltaKey& key, const Slice& deltas, bool* continue_visit) { - if (IsUndoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) { + if (IsUndoRelevant(dfi->opts_.snap, key.timestamp(), continue_visit)) { return ApplyDelete(key, deltas); } return Status::OK(); @@ -846,7 +843,7 @@ template<> inline Status CollectingVisitor<REDO>::Visit(const DeltaKey& key, const Slice& deltas, bool* continue_visit) { - if (IsRedoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) { + if (IsRedoRelevant(dfi->opts_.snap, key.timestamp(), continue_visit)) { return Collect(key, deltas); } return Status::OK(); @@ -856,7 +853,7 @@ template<> inline Status CollectingVisitor<UNDO>::Visit(const DeltaKey& key, const Slice& deltas, bool* continue_visit) { - if (IsUndoRelevant(dfi->mvcc_snap_, key.timestamp(), continue_visit)) { + if (IsUndoRelevant(dfi->opts_.snap, key.timestamp(), continue_visit)) { return Collect(key, deltas); } return Status::OK(); @@ -939,7 +936,7 @@ Status DeltaFileIterator::FilterColumnIdsAndCollectDeltas( void DeltaFileIterator::FatalUnexpectedDelta(const DeltaKey &key, const Slice &deltas, const string &msg) { LOG(FATAL) << "Saw unexpected delta type in deltafile " << dfr_->ToString() << ": " - << " rcl=" << RowChangeList(deltas).ToString(*projection_) + << " rcl=" << RowChangeList(deltas).ToString(*opts_.projection) << " key=" << key.ToString() << " (" << msg << ")"; } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/deltafile.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/deltafile.h b/src/kudu/tablet/deltafile.h index d63af84..e1e418f 100644 --- a/src/kudu/tablet/deltafile.h +++ b/src/kudu/tablet/deltafile.h @@ -40,7 +40,7 @@ #include "kudu/tablet/delta_key.h" #include "kudu/tablet/delta_stats.h" #include "kudu/tablet/delta_store.h" -#include "kudu/tablet/mvcc.h" +#include "kudu/tablet/rowset.h" #include "kudu/util/faststring.h" #include "kudu/util/once.h" #include "kudu/util/slice.h" @@ -55,10 +55,13 @@ class FsManager; class MemTracker; class RowChangeList; class ScanSpec; -class Schema; class SelectionVector; struct ColumnId; +namespace tablet { +class MvccSnapshot; +} // namespace tablet + namespace cfile { struct ReaderOptions; } // namespace cfile @@ -162,8 +165,7 @@ class DeltaFileReader : public DeltaStore, } // See DeltaStore::NewDeltaIterator(...) - Status NewDeltaIterator(const Schema *projection, - const MvccSnapshot &snap, + Status NewDeltaIterator(const RowIteratorOptions& opts, DeltaIterator** iterator) const OVERRIDE; // See DeltaStore::CheckRowDeleted @@ -288,10 +290,9 @@ class DeltaFileIterator : public DeltaIterator { }; - // The passed 'projection' and 'dfr' must remain valid for the lifetime - // of the iterator. + // The pointers in 'opts' and 'dfr' must remain valid for the lifetime of the iterator. DeltaFileIterator(std::shared_ptr<DeltaFileReader> dfr, - const Schema *projection, MvccSnapshot snap, + RowIteratorOptions opts, DeltaType delta_type); // Determine the row index of the first update in the block currently @@ -317,11 +318,7 @@ class DeltaFileIterator : public DeltaIterator { std::shared_ptr<DeltaFileReader> dfr_; - // Schema used during projection. - const Schema* projection_; - - // The MVCC state which determines which deltas should be applied. - const MvccSnapshot mvcc_snap_; + const RowIteratorOptions opts_; gscoped_ptr<cfile::IndexTreeIterator> index_iter_; http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/deltamemstore-test.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/deltamemstore-test.cc b/src/kudu/tablet/deltamemstore-test.cc index 8f17b34..e60cdc6 100644 --- a/src/kudu/tablet/deltamemstore-test.cc +++ b/src/kudu/tablet/deltamemstore-test.cc @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +#include "kudu/tablet/deltamemstore.h" + #include <algorithm> #include <cstdint> #include <cstdio> @@ -51,9 +53,9 @@ #include "kudu/tablet/delta_stats.h" #include "kudu/tablet/delta_store.h" #include "kudu/tablet/deltafile.h" -#include "kudu/tablet/deltamemstore.h" #include "kudu/tablet/mutation.h" #include "kudu/tablet/mvcc.h" +#include "kudu/tablet/rowset.h" #include "kudu/util/faststring.h" #include "kudu/util/mem_tracker.h" #include "kudu/util/memory/arena.h" @@ -130,9 +132,11 @@ class TestDeltaMemStore : public KuduTest { Schema single_col_projection({ col_schema }, { schema_.column_id(col_idx) }, 0); - + RowIteratorOptions opts; + opts.projection = &single_col_projection; + opts.snap = snapshot; DeltaIterator* raw_iter; - Status s = dms_->NewDeltaIterator(&single_col_projection, snapshot, &raw_iter); + Status s = dms_->NewDeltaIterator(opts, &raw_iter); if (s.IsNotFound()) { return; } @@ -443,12 +447,14 @@ TEST_F(TestDeltaMemStore, TestIteratorDoesUpdates) { UpdateIntsAtIndexes(to_update); ASSERT_EQ(1000, dms_->Count()); - // TODO: test snapshot reads from different points - MvccSnapshot snap(mvcc_); ScopedColumnBlock<UINT32> block(100); + RowIteratorOptions opts; + opts.projection = &schema_; + // TODO(todd): test snapshot reads from different points + opts.snap = MvccSnapshot(mvcc_); DeltaIterator* raw_iter; - Status s = dms_->NewDeltaIterator(&schema_, snap, &raw_iter); + Status s = dms_->NewDeltaIterator(opts, &raw_iter); if (s.IsNotFound()) { FAIL() << "Iterator fell outside of the range of the snapshot"; } @@ -488,14 +494,15 @@ TEST_F(TestDeltaMemStore, TestCollectMutations) { ASSERT_EQ(2, dms_->Count()); - MvccSnapshot snap(mvcc_); - const int kBatchSize = 10; vector<Mutation *> mutations; mutations.resize(kBatchSize); + RowIteratorOptions opts; + opts.projection = &schema_; + opts.snap = MvccSnapshot(mvcc_); DeltaIterator* raw_iter; - Status s = dms_->NewDeltaIterator(&schema_, snap, &raw_iter); + Status s = dms_->NewDeltaIterator(opts, &raw_iter); if (s.IsNotFound()) { FAIL() << "Iterator fell outside of the range of the snapshot"; } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/deltamemstore.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/deltamemstore.cc b/src/kudu/tablet/deltamemstore.cc index bed6dfd..c1abcef 100644 --- a/src/kudu/tablet/deltamemstore.cc +++ b/src/kudu/tablet/deltamemstore.cc @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +#include "kudu/tablet/deltamemstore.h" + #include <cstring> #include <ostream> #include <utility> @@ -32,10 +34,10 @@ #include "kudu/gutil/port.h" #include "kudu/gutil/strings/substitute.h" #include "kudu/tablet/deltafile.h" -#include "kudu/tablet/deltamemstore.h" #include "kudu/tablet/mutation.h" #include "kudu/tablet/mvcc.h" #include "kudu/util/debug-util.h" +#include "kudu/util/faststring.h" #include "kudu/util/memcmpable_varint.h" #include "kudu/util/memory/memory.h" #include "kudu/util/status.h" @@ -143,10 +145,9 @@ Status DeltaMemStore::FlushToFile(DeltaFileWriter *dfw, return Status::OK(); } -Status DeltaMemStore::NewDeltaIterator(const Schema *projection, - const MvccSnapshot &snap, +Status DeltaMemStore::NewDeltaIterator(const RowIteratorOptions& opts, DeltaIterator** iterator) const { - *iterator = new DMSIterator(shared_from_this(), projection, snap); + *iterator = new DMSIterator(shared_from_this(), opts); return Status::OK(); } @@ -195,18 +196,17 @@ void DeltaMemStore::DebugPrint() const { //////////////////////////////////////////////////////////// DMSIterator::DMSIterator(const shared_ptr<const DeltaMemStore>& dms, - const Schema* projection, MvccSnapshot snapshot) + RowIteratorOptions opts) : dms_(dms), - mvcc_snapshot_(std::move(snapshot)), + opts_(std::move(opts)), iter_(dms->tree_.NewIterator()), initted_(false), prepared_idx_(0), prepared_count_(0), prepared_for_(NOT_PREPARED), - seeked_(false), - projection_(projection) {} + seeked_(false) {} -Status DMSIterator::Init(ScanSpec *spec) { +Status DMSIterator::Init(ScanSpec* /*spec*/) { initted_ = true; return Status::OK(); } @@ -243,7 +243,7 @@ Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) { rowid_t stop_row = start_row + nrows - 1; if (updates_by_col_.empty()) { - updates_by_col_.resize(projection_->num_columns()); + updates_by_col_.resize(opts_.projection->num_columns()); } for (UpdatesForColumn& ufc : updates_by_col_) { ufc.clear(); @@ -259,7 +259,7 @@ Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) { DCHECK_GE(key.row_idx(), start_row); if (key.row_idx() > stop_row) break; - if (!mvcc_snapshot_.IsCommitted(key.timestamp())) { + if (!opts_.snap.IsCommitted(key.timestamp())) { // The transaction which applied this update is not yet committed // in this iterator's MVCC snapshot. Hence, skip it. iter_->Next(); @@ -279,12 +279,12 @@ Status DMSIterator::PrepareBatch(size_t nrows, PrepareFlag flag) { RETURN_NOT_OK(decoder.DecodeNext(&dec)); int col_idx; const void* col_val; - RETURN_NOT_OK(dec.Validate(*projection_, &col_idx, &col_val)); + RETURN_NOT_OK(dec.Validate(*opts_.projection, &col_idx, &col_val)); if (col_idx == -1) { // This column isn't being projected. continue; } - int col_size = projection_->column(col_idx).type_info()->size(); + int col_size = opts_.projection->column(col_idx).type_info()->size(); // If we already have an earlier update for the same column, we can // just overwrite that one. @@ -325,7 +325,7 @@ Status DMSIterator::ApplyUpdates(size_t col_to_apply, ColumnBlock *dst) { DCHECK_EQ(prepared_for_, PREPARED_FOR_APPLY); DCHECK_EQ(prepared_count_, dst->nrows()); - const ColumnSchema* col_schema = &projection_->column(col_to_apply); + const ColumnSchema* col_schema = &opts_.projection->column(col_to_apply); for (const ColumnUpdate& cu : updates_by_col_[col_to_apply]) { int32_t idx_in_block = cu.row_id - prepared_idx_; DCHECK_GE(idx_in_block, 0); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/deltamemstore.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/deltamemstore.h b/src/kudu/tablet/deltamemstore.h index a1d2994..b637e00 100644 --- a/src/kudu/tablet/deltamemstore.h +++ b/src/kudu/tablet/deltamemstore.h @@ -36,9 +36,8 @@ #include "kudu/tablet/delta_key.h" #include "kudu/tablet/delta_stats.h" #include "kudu/tablet/delta_store.h" -#include "kudu/tablet/mvcc.h" +#include "kudu/tablet/rowset.h" #include "kudu/util/atomic.h" -#include "kudu/util/faststring.h" #include "kudu/util/memory/arena.h" #include "kudu/util/slice.h" #include "kudu/util/status.h" @@ -50,7 +49,6 @@ class MemTracker; class MemoryTrackingBufferAllocator; class RowChangeList; class ScanSpec; -class Schema; class SelectionVector; class Timestamp; struct ColumnId; @@ -112,17 +110,16 @@ class DeltaMemStore : public DeltaStore, // Create an iterator for applying deltas from this DMS. // - // The projection passed here must be the same as the schema of any + // The projection passed in 'opts' must be the same as the schema of any // RowBlocks which are passed in, or else bad things will happen. // - // 'snapshot' is the MVCC state which determines which transactions + // The snapshot in 'opts' is the MVCC state which determines which transactions // should be considered committed (and thus applied by the iterator). // // Returns Status::OK and sets 'iterator' to the new DeltaIterator, or // returns Status::NotFound if the mutations within this delta store - // cannot include 'snap'. - virtual Status NewDeltaIterator(const Schema *projection, - const MvccSnapshot &snap, + // cannot include the snapshot. + virtual Status NewDeltaIterator(const RowIteratorOptions& opts, DeltaIterator** iterator) const OVERRIDE; virtual Status CheckRowDeleted(rowid_t row_idx, bool *deleted) const OVERRIDE; @@ -229,14 +226,13 @@ class DMSIterator : public DeltaIterator { // Initialize the iterator. // The projection passed here must be the same as the schema of any // RowBlocks which are passed in, or else bad things will happen. - // The pointer must also remain valid for the lifetime of the iterator. + // The pointers in 'opts' must also remain valid for the lifetime of the iterator. DMSIterator(const std::shared_ptr<const DeltaMemStore> &dms, - const Schema *projection, MvccSnapshot snapshot); + RowIteratorOptions opts); const std::shared_ptr<const DeltaMemStore> dms_; - // MVCC state which allows us to ignore uncommitted transactions. - const MvccSnapshot mvcc_snapshot_; + const RowIteratorOptions opts_; gscoped_ptr<DeltaMemStore::DMSTreeIter> iter_; @@ -259,9 +255,6 @@ class DMSIterator : public DeltaIterator { // True if SeekToOrdinal() been called at least once. bool seeked_; - // The schema of the row blocks that will be passed to PrepareBatch(), etc. - const Schema* projection_; - // State when prepared_for_ == PREPARED_FOR_APPLY // ------------------------------------------------------------ struct ColumnUpdate { http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/diskrowset-test-base.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/diskrowset-test-base.h b/src/kudu/tablet/diskrowset-test-base.h index cbe5c26..9742169 100644 --- a/src/kudu/tablet/diskrowset-test-base.h +++ b/src/kudu/tablet/diskrowset-test-base.h @@ -212,10 +212,11 @@ class TestRowSet : public KuduRowSetTest { void VerifyUpdatesWithRowIter(const DiskRowSet &rs, const std::unordered_set<uint32_t> &updated) { Schema proj_val = CreateProjection(schema_, { "val" }); - MvccSnapshot snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions(); + RowIteratorOptions opts; + opts.projection = &proj_val; gscoped_ptr<RowwiseIterator> row_iter; - CHECK_OK(rs.NewRowIterator(&proj_val, snap, UNORDERED, &row_iter)); - CHECK_OK(row_iter->Init(NULL)); + CHECK_OK(rs.NewRowIterator(opts, &row_iter)); + CHECK_OK(row_iter->Init(nullptr)); Arena arena(1024); int batch_size = 10000; RowBlock dst(proj_val, batch_size, &arena); @@ -259,9 +260,10 @@ class TestRowSet : public KuduRowSetTest { spec.AddPredicate(pred); spec.OptimizeScan(schema_, &arena, &pool, true); - MvccSnapshot snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions(); + RowIteratorOptions opts; + opts.projection = &schema_; gscoped_ptr<RowwiseIterator> row_iter; - CHECK_OK(rs.NewRowIterator(&schema_, snap, UNORDERED, &row_iter)); + CHECK_OK(rs.NewRowIterator(opts, &row_iter)); CHECK_OK(row_iter->Init(&spec)); std::vector<std::string> rows; IterateToStringList(row_iter.get(), &rows); @@ -273,10 +275,11 @@ class TestRowSet : public KuduRowSetTest { // using the given schema as a projection. static void IterateProjection(const DiskRowSet &rs, const Schema &schema, int expected_rows, bool do_log = true) { - MvccSnapshot snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions(); + RowIteratorOptions opts; + opts.projection = &schema; gscoped_ptr<RowwiseIterator> row_iter; - CHECK_OK(rs.NewRowIterator(&schema, snap, UNORDERED, &row_iter)); - CHECK_OK(row_iter->Init(NULL)); + CHECK_OK(rs.NewRowIterator(opts, &row_iter)); + CHECK_OK(row_iter->Init(nullptr)); int batch_size = 1000; Arena arena(1024); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/diskrowset-test.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/diskrowset-test.cc b/src/kudu/tablet/diskrowset-test.cc index d2c6867..e64cdfa 100644 --- a/src/kudu/tablet/diskrowset-test.cc +++ b/src/kudu/tablet/diskrowset-test.cc @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +#include "kudu/tablet/diskrowset.h" + #include <algorithm> #include <cstdint> #include <memory> @@ -30,7 +32,6 @@ #include <gtest/gtest.h> #include "kudu/clock/clock.h" -#include "kudu/common/common.pb.h" #include "kudu/common/iterator.h" #include "kudu/common/row.h" #include "kudu/common/row_changelist.h" @@ -47,7 +48,6 @@ #include "kudu/tablet/delta_tracker.h" #include "kudu/tablet/deltamemstore.h" #include "kudu/tablet/diskrowset-test-base.h" -#include "kudu/tablet/diskrowset.h" #include "kudu/tablet/mvcc.h" #include "kudu/tablet/rowset.h" #include "kudu/tablet/rowset_metadata.h" @@ -414,8 +414,11 @@ TEST_F(TestRowSet, TestFlushedUpdatesRespectMVCC) { ASSERT_EQ(5, snaps.size()); for (int i = 0; i < 5; i++) { SCOPED_TRACE(i); + RowIteratorOptions opts; + opts.projection = &schema_; + opts.snap = snaps[i]; gscoped_ptr<RowwiseIterator> iter; - ASSERT_OK(rs->NewRowIterator(&schema_, snaps[i], UNORDERED, &iter)); + ASSERT_OK(rs->NewRowIterator(opts, &iter)); string data = InitAndDumpIterator(std::move(iter)); EXPECT_EQ(StringPrintf(R"((string key="row", uint32 val=%d))", i + 1), data); } @@ -426,8 +429,11 @@ TEST_F(TestRowSet, TestFlushedUpdatesRespectMVCC) { for (int i = 0; i < 5; i++) { SCOPED_TRACE(i); + RowIteratorOptions opts; + opts.projection = &schema_; + opts.snap = snaps[i]; gscoped_ptr<RowwiseIterator> iter; - ASSERT_OK(rs->NewRowIterator(&schema_, snaps[i], UNORDERED, &iter)); + ASSERT_OK(rs->NewRowIterator(opts, &iter)); string data = InitAndDumpIterator(std::move(iter)); EXPECT_EQ(StringPrintf(R"((string key="row", uint32 val=%d))", i + 1), data); } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/diskrowset.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/diskrowset.cc b/src/kudu/tablet/diskrowset.cc index 29685f4..6517fdb 100644 --- a/src/kudu/tablet/diskrowset.cc +++ b/src/kudu/tablet/diskrowset.cc @@ -30,6 +30,7 @@ #include "kudu/cfile/bloomfile.h" #include "kudu/cfile/cfile_util.h" #include "kudu/cfile/cfile_writer.h" +#include "kudu/common/common.pb.h" #include "kudu/common/generic_iterators.h" #include "kudu/common/iterator.h" #include "kudu/common/rowblock.h" @@ -626,14 +627,12 @@ Status DiskRowSet::NewMajorDeltaCompaction(const vector<ColumnId>& col_ids, const Schema* schema = &rowset_metadata_->tablet_schema(); - vector<shared_ptr<DeltaStore> > included_stores; + RowIteratorOptions opts; + opts.projection = schema; + vector<shared_ptr<DeltaStore>> included_stores; unique_ptr<DeltaIterator> delta_iter; RETURN_NOT_OK(delta_tracker_->NewDeltaFileIterator( - schema, - MvccSnapshot::CreateSnapshotIncludingAllTransactions(), - REDO, - &included_stores, - &delta_iter)); + opts, REDO, &included_stores, &delta_iter)); out->reset(new MajorDeltaCompaction(rowset_metadata_->fs_manager(), *schema, @@ -646,16 +645,14 @@ Status DiskRowSet::NewMajorDeltaCompaction(const vector<ColumnId>& col_ids, return Status::OK(); } -Status DiskRowSet::NewRowIterator(const Schema *projection, - const MvccSnapshot &mvcc_snap, - OrderMode /*order*/, +Status DiskRowSet::NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const { DCHECK(open_); shared_lock<rw_spinlock> l(component_lock_); - shared_ptr<CFileSet::Iterator> base_iter(base_data_->NewIterator(projection)); + shared_ptr<CFileSet::Iterator> base_iter(base_data_->NewIterator(opts.projection)); gscoped_ptr<ColumnwiseIterator> col_iter; - RETURN_NOT_OK(delta_tracker_->WrapIterator(base_iter, mvcc_snap, &col_iter)); + RETURN_NOT_OK(delta_tracker_->WrapIterator(base_iter, opts, &col_iter)); out->reset(new MaterializingIterator( shared_ptr<ColumnwiseIterator>(col_iter.release()))); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/diskrowset.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/diskrowset.h b/src/kudu/tablet/diskrowset.h index ba1b325..67e587c 100644 --- a/src/kudu/tablet/diskrowset.h +++ b/src/kudu/tablet/diskrowset.h @@ -31,7 +31,6 @@ #include <glog/logging.h> #include <gtest/gtest_prod.h> -#include "kudu/common/common.pb.h" #include "kudu/common/rowid.h" #include "kudu/common/schema.h" #include "kudu/fs/block_id.h" @@ -352,9 +351,7 @@ class DiskRowSet : public RowSet { //////////////////// // Read functions. //////////////////// - virtual Status NewRowIterator(const Schema *projection, - const MvccSnapshot &mvcc_snap, - OrderMode order, + virtual Status NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const override; virtual Status NewCompactionInput(const Schema* projection, http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/memrowset-test.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/memrowset-test.cc b/src/kudu/tablet/memrowset-test.cc index 6d7aaf3..047039f 100644 --- a/src/kudu/tablet/memrowset-test.cc +++ b/src/kudu/tablet/memrowset-test.cc @@ -15,6 +15,8 @@ // specific language governing permissions and limitations // under the License. +#include "kudu/tablet/memrowset.h" + #include <cstdint> #include <cstdio> #include <memory> @@ -40,7 +42,6 @@ #include "kudu/gutil/gscoped_ptr.h" #include "kudu/gutil/ref_counted.h" #include "kudu/gutil/stringprintf.h" -#include "kudu/tablet/memrowset.h" #include "kudu/tablet/mvcc.h" #include "kudu/tablet/rowset.h" #include "kudu/tablet/tablet-test-util.h" @@ -189,8 +190,11 @@ class TestMemRowSet : public KuduTest { } int ScanAndCount(MemRowSet* mrs, const MvccSnapshot& snap) { - gscoped_ptr<MemRowSet::Iterator> iter(mrs->NewIterator(&schema_, snap)); - CHECK_OK(iter->Init(NULL)); + RowIteratorOptions opts; + opts.projection = &schema_; + opts.snap = snap; + gscoped_ptr<MemRowSet::Iterator> iter(mrs->NewIterator(opts)); + CHECK_OK(iter->Init(nullptr)); Arena arena(1024); RowBlock block(schema_, 100, &arena); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/memrowset.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc index 81acc3a..3d25bc7 100644 --- a/src/kudu/tablet/memrowset.cc +++ b/src/kudu/tablet/memrowset.cc @@ -29,7 +29,6 @@ #include "kudu/codegen/compilation_manager.h" #include "kudu/codegen/row_projector.h" #include "kudu/common/columnblock.h" -#include "kudu/common/common.pb.h" #include "kudu/common/encoded_key.h" #include "kudu/common/row.h" #include "kudu/common/row_changelist.h" @@ -42,6 +41,7 @@ #include "kudu/gutil/strings/substitute.h" #include "kudu/tablet/compaction.h" #include "kudu/tablet/mutation.h" +#include "kudu/tablet/mvcc.h" #include "kudu/tablet/tablet.pb.h" #include "kudu/util/flag_tags.h" #include "kudu/util/mem_tracker.h" @@ -286,22 +286,20 @@ Status MemRowSet::CheckRowPresent(const RowSetKeyProbe &probe, bool *present, return Status::OK(); } -MemRowSet::Iterator *MemRowSet::NewIterator(const Schema *projection, - const MvccSnapshot &snap) const { - return new MemRowSet::Iterator(shared_from_this(), tree_.NewIterator(), - projection, snap); +MemRowSet::Iterator *MemRowSet::NewIterator(const RowIteratorOptions& opts) const { + return new MemRowSet::Iterator(shared_from_this(), tree_.NewIterator(), opts); } MemRowSet::Iterator *MemRowSet::NewIterator() const { - // TODO: can we kill this function? should be only used by tests? - return NewIterator(&schema(), MvccSnapshot::CreateSnapshotIncludingAllTransactions()); + // TODO(todd): can we kill this function? should be only used by tests? + RowIteratorOptions opts; + opts.projection = &schema(); + return NewIterator(opts); } -Status MemRowSet::NewRowIterator(const Schema *projection, - const MvccSnapshot &snap, - OrderMode /*order*/, +Status MemRowSet::NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const { - out->reset(NewIterator(projection, snap)); + out->reset(NewIterator(opts)); return Status::OK(); } @@ -386,16 +384,15 @@ gscoped_ptr<MRSRowProjector> GenerateAppropriateProjector( MemRowSet::Iterator::Iterator(const std::shared_ptr<const MemRowSet>& mrs, MemRowSet::MSBTIter* iter, - const Schema* projection, MvccSnapshot mvcc_snap) + RowIteratorOptions opts) : memrowset_(mrs), iter_(iter), - mvcc_snap_(std::move(mvcc_snap)), - projection_(projection), + opts_(std::move(opts)), projector_( - GenerateAppropriateProjector(&mrs->schema_nonvirtual(), projection)), - delta_projector_(&mrs->schema_nonvirtual(), projection), + GenerateAppropriateProjector(&mrs->schema_nonvirtual(), opts_.projection)), + delta_projector_(&mrs->schema_nonvirtual(), opts_.projection), state_(kUninitialized) { - // TODO: various code assumes that a newly constructed iterator + // TODO(todd): various code assumes that a newly constructed iterator // is pointed at the beginning of the dataset. This causes a redundant // seek. Could make this lazy instead, or change the semantics so that // a seek is required (probably the latter) @@ -499,7 +496,7 @@ Status MemRowSet::Iterator::FetchRows(RowBlock* dst, size_t* fetched) { iter_->GetCurrentEntry(&k, &v); MRSRow row(memrowset_.get(), v); - if (mvcc_snap_.IsCommitted(row.insertion_timestamp())) { + if (opts_.snap.IsCommitted(row.insertion_timestamp())) { if (has_upper_bound() && out_of_bounds(k)) { state_ = kFinished; break; @@ -545,7 +542,7 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow( for (const Mutation *mut = mutation_head; mut != nullptr; mut = mut->acquire_next()) { - if (!mvcc_snap_.IsCommitted(mut->timestamp_)) { + if (!opts_.snap.IsCommitted(mut->timestamp_)) { // Transaction which wasn't committed yet in the reader's snapshot. continue; } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/memrowset.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/memrowset.h b/src/kudu/tablet/memrowset.h index 5d83a91..aba1988 100644 --- a/src/kudu/tablet/memrowset.h +++ b/src/kudu/tablet/memrowset.h @@ -28,7 +28,6 @@ #include <boost/optional/optional.hpp> #include <glog/logging.h> -#include "kudu/common/common.pb.h" #include "kudu/common/iterator.h" #include "kudu/common/row.h" #include "kudu/common/rowid.h" @@ -39,7 +38,6 @@ #include "kudu/gutil/gscoped_ptr.h" #include "kudu/gutil/macros.h" #include "kudu/tablet/concurrent_btree.h" -#include "kudu/tablet/mvcc.h" #include "kudu/tablet/rowset.h" #include "kudu/tablet/rowset_metadata.h" #include "kudu/util/faststring.h" @@ -58,9 +56,13 @@ class RowChangeList; class ScanSpec; struct IteratorStats; +namespace tablet { +class MvccSnapshot; +} // namespace tablet + namespace consensus { class OpId; -} +} // namespace consensus namespace tablet { // @@ -309,13 +311,10 @@ class MemRowSet : public RowSet, // // TODO(todd): clarify the consistency of this iterator in the method doc Iterator *NewIterator() const; - Iterator *NewIterator(const Schema *projection, - const MvccSnapshot &snap) const; + Iterator *NewIterator(const RowIteratorOptions& opts) const; // Alias to conform to DiskRowSet interface - virtual Status NewRowIterator(const Schema* projection, - const MvccSnapshot& snap, - OrderMode order, + virtual Status NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const override; // Create compaction input. @@ -515,7 +514,7 @@ class MemRowSet::Iterator : public RowwiseIterator { } const Schema& schema() const override { - return *projection_; + return *opts_.projection; } virtual void GetIteratorStats(std::vector<IteratorStats>* stats) const override { @@ -541,8 +540,7 @@ class MemRowSet::Iterator : public RowwiseIterator { DISALLOW_COPY_AND_ASSIGN(Iterator); Iterator(const std::shared_ptr<const MemRowSet> &mrs, - MemRowSet::MSBTIter *iter, const Schema *projection, - MvccSnapshot mvcc_snap); + MemRowSet::MSBTIter *iter, RowIteratorOptions opts); // Various helper functions called while getting the next RowBlock Status FetchRows(RowBlock* dst, size_t* fetched); @@ -553,15 +551,12 @@ class MemRowSet::Iterator : public RowwiseIterator { const std::shared_ptr<const MemRowSet> memrowset_; gscoped_ptr<MemRowSet::MSBTIter> iter_; - // The MVCC snapshot which determines which rows and mutations are visible to - // this iterator. - const MvccSnapshot mvcc_snap_; + RowIteratorOptions opts_; // Mapping from projected column index back to memrowset column index. // Relies on the MRSRowProjector interface to abstract from the two // different implementations of the RowProjector, which may change // at runtime (using vs. not using code generation). - const Schema* const projection_; gscoped_ptr<MRSRowProjector> projector_; DeltaProjector delta_projector_; http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/mock-rowsets.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/mock-rowsets.h b/src/kudu/tablet/mock-rowsets.h index 51763d6..b83fbc5 100644 --- a/src/kudu/tablet/mock-rowsets.h +++ b/src/kudu/tablet/mock-rowsets.h @@ -47,9 +47,7 @@ class MockRowSet : public RowSet { LOG(FATAL) << "Unimplemented"; return Status::OK(); } - virtual Status NewRowIterator(const Schema* /*projection*/, - const MvccSnapshot& /*snap*/, - OrderMode /*order*/, + virtual Status NewRowIterator(const RowIteratorOptions& /*opts*/, gscoped_ptr<RowwiseIterator>* /*out*/) const OVERRIDE { LOG(FATAL) << "Unimplemented"; return Status::OK(); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/mt-rowset_delta_compaction-test.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/mt-rowset_delta_compaction-test.cc b/src/kudu/tablet/mt-rowset_delta_compaction-test.cc index 776df6a..ad1acab 100644 --- a/src/kudu/tablet/mt-rowset_delta_compaction-test.cc +++ b/src/kudu/tablet/mt-rowset_delta_compaction-test.cc @@ -34,7 +34,7 @@ #include "kudu/gutil/strings/substitute.h" #include "kudu/tablet/diskrowset-test-base.h" #include "kudu/tablet/diskrowset.h" -#include "kudu/tablet/mvcc.h" +#include "kudu/tablet/rowset.h" #include "kudu/tablet/tablet.pb.h" #include "kudu/util/memory/arena.h" #include "kudu/util/monotime.h" @@ -109,11 +109,10 @@ class TestMultiThreadedRowSetDeltaCompaction : public TestRowSet { void ReadVerify(DiskRowSet *rs) { Arena arena(1024); RowBlock dst(schema_, 1000, &arena); + RowIteratorOptions opts; + opts.projection = &schema_; gscoped_ptr<RowwiseIterator> iter; - ASSERT_OK(rs->NewRowIterator(&schema_, - MvccSnapshot::CreateSnapshotIncludingAllTransactions(), - UNORDERED, - &iter)); + ASSERT_OK(rs->NewRowIterator(opts, &iter)); uint32_t expected = NoBarrier_Load(&update_counter_); ASSERT_OK(iter->Init(nullptr)); while (iter->HasNext()) { http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/rowset.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/rowset.cc b/src/kudu/tablet/rowset.cc index 7122aad..cf060f7 100644 --- a/src/kudu/tablet/rowset.cc +++ b/src/kudu/tablet/rowset.cc @@ -35,7 +35,13 @@ using std::string; using std::vector; using strings::Substitute; -namespace kudu { namespace tablet { +namespace kudu { +namespace tablet { + +RowIteratorOptions::RowIteratorOptions() + : projection(nullptr), + snap(MvccSnapshot::CreateSnapshotIncludingAllTransactions()), + order(OrderMode::UNORDERED) {} DuplicatingRowSet::DuplicatingRowSet(RowSetVector old_rowsets, RowSetVector new_rowsets) @@ -73,34 +79,32 @@ string DuplicatingRowSet::ToString() const { return ret; } -Status DuplicatingRowSet::NewRowIterator(const Schema *projection, - const MvccSnapshot &snap, - OrderMode order, +Status DuplicatingRowSet::NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const { // Use the original rowset. if (old_rowsets_.size() == 1) { - return old_rowsets_[0]->NewRowIterator(projection, snap, order, out); + return old_rowsets_[0]->NewRowIterator(opts, out); } // Union or merge between them vector<shared_ptr<RowwiseIterator> > iters; for (const shared_ptr<RowSet> &rowset : old_rowsets_) { gscoped_ptr<RowwiseIterator> iter; - RETURN_NOT_OK_PREPEND(rowset->NewRowIterator(projection, snap, order, &iter), + RETURN_NOT_OK_PREPEND(rowset->NewRowIterator(opts, &iter), Substitute("Could not create iterator for rowset $0", rowset->ToString())); iters.push_back(shared_ptr<RowwiseIterator>(iter.release())); } - switch (order) { + switch (opts.order) { case ORDERED: - out->reset(new MergeIterator(*projection, std::move(iters))); + out->reset(new MergeIterator(*opts.projection, std::move(iters))); break; case UNORDERED: out->reset(new UnionIterator(std::move(iters))); break; default: - LOG(FATAL) << "unknown order: " << order; + LOG(FATAL) << "unknown order: " << opts.order; } return Status::OK(); } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/rowset.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/rowset.h b/src/kudu/tablet/rowset.h index 89ef088..3dcf920 100644 --- a/src/kudu/tablet/rowset.h +++ b/src/kudu/tablet/rowset.h @@ -35,6 +35,7 @@ #include "kudu/gutil/gscoped_ptr.h" #include "kudu/gutil/macros.h" #include "kudu/gutil/port.h" +#include "kudu/tablet/mvcc.h" #include "kudu/util/bloom_filter.h" #include "kudu/util/status.h" // IWYU pragma: no_include "kudu/util/monotime.h" @@ -55,11 +56,31 @@ namespace tablet { class CompactionInput; class OperationResultPB; -class MvccSnapshot; class RowSetKeyProbe; class RowSetMetadata; struct ProbeStats; +// Encapsulates all options passed to row-based Iterators. +struct RowIteratorOptions { + RowIteratorOptions(); + + // The projection to use in the iteration. + // + // Defaults to nullptr. + const Schema* projection; + + // Transactions not committed in this snapshot will be ignored in the iteration. + // + // Defaults to a snapshot that includes all transactions. + MvccSnapshot snap; + + // Whether iteration should be ordered by primary key. Only relevant to those + // iterators that deal with primary key order. + // + // Defaults to UNORDERED. + OrderMode order; +}; + class RowSet { public: enum DeltaCompactionType { @@ -89,17 +110,19 @@ class RowSet { ProbeStats* stats, OperationResultPB* result) = 0; - // Return a new RowIterator for this rowset, with the given projection. - // The projection schema must remain valid for the lifetime of the iterator. + // Return a new RowIterator for this rowset, with the given options. + // + // Pointers in 'opts' must remain valid for the lifetime of the iterator. + // // The iterator will return rows/updates which were committed as of the time of - // 'snap'. + // the snapshot in 'opts'. + // // The returned iterator is not Initted. - virtual Status NewRowIterator(const Schema *projection, - const MvccSnapshot &snap, - OrderMode order, + virtual Status NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const = 0; // Create the input to be used for a compaction. + // // The provided 'projection' is for the compaction output. Each row // will be projected into this Schema. virtual Status NewCompactionInput(const Schema* projection, @@ -336,9 +359,7 @@ class DuplicatingRowSet : public RowSet { Status CheckRowPresent(const RowSetKeyProbe &probe, bool *present, ProbeStats* stats) const OVERRIDE; - virtual Status NewRowIterator(const Schema *projection, - const MvccSnapshot &snap, - OrderMode order, + virtual Status NewRowIterator(const RowIteratorOptions& opts, gscoped_ptr<RowwiseIterator>* out) const OVERRIDE; virtual Status NewCompactionInput(const Schema* projection, http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/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 1881df8..e43707d 100644 --- a/src/kudu/tablet/tablet-test-util.h +++ b/src/kudu/tablet/tablet-test-util.h @@ -233,9 +233,12 @@ static inline Status DumpRowSet(const RowSet &rs, const MvccSnapshot &snap, std::vector<std::string> *out, int limit = INT_MAX) { + RowIteratorOptions opts; + opts.projection = &projection; + opts.snap = snap; gscoped_ptr<RowwiseIterator> iter; - RETURN_NOT_OK(rs.NewRowIterator(&projection, snap, UNORDERED, &iter)); - RETURN_NOT_OK(iter->Init(NULL)); + RETURN_NOT_OK(rs.NewRowIterator(opts, &iter)); + RETURN_NOT_OK(iter->Init(nullptr)); RETURN_NOT_OK(IterateToStringList(iter.get(), out, limit)); return Status::OK(); } http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/tablet.cc ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc index 9ea89db..4a59847 100644 --- a/src/kudu/tablet/tablet.cc +++ b/src/kudu/tablet/tablet.cc @@ -1737,9 +1737,14 @@ Status Tablet::CaptureConsistentIterators( // in the middle, we don't modify the output arguments. vector<shared_ptr<RowwiseIterator>> ret; + RowIteratorOptions opts; + opts.projection = projection; + opts.snap = snap; + opts.order = order; + // Grab the memrowset iterator. gscoped_ptr<RowwiseIterator> ms_iter; - RETURN_NOT_OK(components_->memrowset->NewRowIterator(projection, snap, order, &ms_iter)); + RETURN_NOT_OK(components_->memrowset->NewRowIterator(opts, &ms_iter)); ret.emplace_back(ms_iter.release()); // Cull row-sets in the case of key-range queries. @@ -1755,7 +1760,7 @@ Status Tablet::CaptureConsistentIterators( &interval_sets); for (const RowSet *rs : interval_sets) { gscoped_ptr<RowwiseIterator> row_it; - RETURN_NOT_OK_PREPEND(rs->NewRowIterator(projection, snap, order, &row_it), + RETURN_NOT_OK_PREPEND(rs->NewRowIterator(opts, &row_it), Substitute("Could not create iterator for rowset $0", rs->ToString())); ret.emplace_back(row_it.release()); @@ -1768,7 +1773,7 @@ Status Tablet::CaptureConsistentIterators( // fall back to grabbing all rowset iterators for (const shared_ptr<RowSet> &rs : components_->rowsets->all_rowsets()) { gscoped_ptr<RowwiseIterator> row_it; - RETURN_NOT_OK_PREPEND(rs->NewRowIterator(projection, snap, order, &row_it), + RETURN_NOT_OK_PREPEND(rs->NewRowIterator(opts, &row_it), Substitute("Could not create iterator for rowset $0", rs->ToString())); ret.emplace_back(row_it.release()); http://git-wip-us.apache.org/repos/asf/kudu/blob/0fbebd67/src/kudu/tablet/tablet.h ---------------------------------------------------------------------- diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h index c17426c..577b5c7 100644 --- a/src/kudu/tablet/tablet.h +++ b/src/kudu/tablet/tablet.h @@ -75,12 +75,12 @@ class AlterSchemaTransactionState; class CompactionPolicy; class HistoryGcOpts; class MemRowSet; -struct RowOp; -class RowSetsInCompaction; class RowSetTree; +class RowSetsInCompaction; +class WriteTransactionState; +struct RowOp; struct TabletComponents; struct TabletMetrics; -class WriteTransactionState; class Tablet { public:
