src/kudu/tablet/rowset.cc - kudu - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.

 #include "kudu/tablet/rowset.h"

 #include <limits>
 #include <memory>
 #include <string>
 #include <type_traits>
 #include <utility>
 #include <vector>

 #include "kudu/common/generic_iterators.h"
 #include "kudu/common/iterator.h"
 #include "kudu/common/schema.h"
 #include "kudu/common/timestamp.h"
 #include "kudu/gutil/strings/substitute.h"
 #include "kudu/tablet/rowset_metadata.h"

 using std::shared_ptr;
 using std::string;
 using std::unique_ptr;
 using std::vector;
 using strings::Substitute;

 namespace kudu {

 using fs::IOContext;

 namespace tablet {

 RowIteratorOptions::RowIteratorOptions()
     : projection(nullptr),
       snap_to_include(MvccSnapshot::CreateSnapshotIncludingAllOps()),
       order(OrderMode::UNORDERED),
       include_deleted_rows(false) {}

 Status RowSet::NewRowIteratorWithBounds(const RowIteratorOptions& opts,
                                         IterWithBounds* out) const {
   // Get the iterator.
   unique_ptr<RowwiseIterator> iter;
   RETURN_NOT_OK(NewRowIterator(opts, &iter));

   // Get the bounds. Some rowsets (e.g. MRS) have no bounds; that's OK as the
   // bounds aren't required.
   string lower;
   string upper;
   Status s = GetBounds(&lower, &upper);
   if (s.ok()) {
     out->encoded_bounds = std::make_pair(std::move(lower), std::move(upper));
   } else if (!s.IsNotSupported()) {
     RETURN_NOT_OK(s);
   }
   out->iter = std::move(iter);
   return Status::OK();
 }

 DuplicatingRowSet::DuplicatingRowSet(RowSetVector old_rowsets,
                                      RowSetVector new_rowsets)
     : old_rowsets_(std::move(old_rowsets)),
       new_rowsets_(std::move(new_rowsets)) {
   CHECK_GT(old_rowsets_.size(), 0);
   CHECK_GT(new_rowsets_.size(), 0);
 }

 DuplicatingRowSet::~DuplicatingRowSet() {
 }

 // Stringify the given list of rowsets into 'dst'.
 static void AppendRowSetStrings(const RowSetVector &rowsets, string *dst) {
   bool first = true;
   dst->append("[");
   for (const shared_ptr<RowSet> &rs : rowsets) {
     if (!first) {
       dst->append(", ");
     }
     first = false;
     dst->append(rs->ToString());
   }
   dst->append("]");
 }

 string DuplicatingRowSet::ToString() const {
   string ret;
   ret.append("DuplicatingRowSet(");

   AppendRowSetStrings(old_rowsets_, &ret);
   ret.append(" -> ");
   AppendRowSetStrings(new_rowsets_, &ret);
   ret.append(")");
   return ret;
 }

 Status DuplicatingRowSet::NewRowIterator(const RowIteratorOptions& opts,
                                          unique_ptr<RowwiseIterator>* out) const {
   // Use the original rowset.
   if (old_rowsets_.size() == 1) {
     return old_rowsets_[0]->NewRowIterator(opts, out);
   }
   // Union or merge between them

   vector<IterWithBounds> iters;
   for (const auto& rs : old_rowsets_) {
     IterWithBounds iwb;
     RETURN_NOT_OK_PREPEND(rs->NewRowIteratorWithBounds(opts, &iwb),
                           Substitute("Could not create iterator for rowset $0",
                                      rs->ToString()));
     iters.emplace_back(std::move(iwb));
   }

   switch (opts.order) {
     case ORDERED:
       *out = NewMergeIterator(MergeIteratorOptions(opts.include_deleted_rows), std::move(iters));
       break;
     case UNORDERED:
       *out = NewUnionIterator(std::move(iters));
       break;
     default:
       LOG(FATAL) << "unknown order: " << opts.order;
   }
   return Status::OK();
 }

 Status DuplicatingRowSet::NewCompactionInput(const Schema* /*projection*/,
                                              const MvccSnapshot& /*snap*/,
                                              const IOContext* /*io_context*/,
                                              unique_ptr<CompactionInput>* /*out*/) const  {
   LOG(FATAL) << "duplicating rowsets do not act as compaction input";
   return Status::OK();
 }

 Status DuplicatingRowSet::MutateRow(Timestamp timestamp,
                                     const RowSetKeyProbe &probe,
                                     const RowChangeList &update,
                                     const consensus::OpId& op_id,
                                     const IOContext* io_context,
                                     ProbeStats* stats,
                                     OperationResultPB* result) {
   // Duplicate the update to both the relevant input rowset and the output rowset.
   //
   // It's crucial to do the mutation against the input side first, due to the potential
   // for a race during flush: the output rowset may not yet hold a DELETE which
   // is present in the input rowset. In that case, the UPDATE against the output rowset would
   // succeed whereas it can't be applied to the input rowset. So, we update the input rowset first,
   // and if it succeeds, propagate to the output.

   // First mutate the relevant input rowset.
   bool updated = false;
   for (const shared_ptr<RowSet> &rowset : old_rowsets_) {
     Status s = rowset->MutateRow(timestamp, probe, update, op_id, io_context, stats, result);
     if (s.ok()) {
       updated = true;
       break;
     } else if (!s.IsNotFound()) {
       LOG(ERROR) << "Unable to update key "
                  << probe.schema()->CreateKeyProjection().DebugRow(probe.row_key())
                  << " (failed on rowset " << rowset->ToString() << "): "
                  << s.ToString();
       return s;
     }
   }

   if (!updated) {
     return Status::NotFound("not found in any compaction input");
   }

   // If it succeeded there, we also need to mirror into the new rowset.
   int mirrored_count = 0;
   for (const shared_ptr<RowSet> &new_rowset : new_rowsets_) {
     Status s = new_rowset->MutateRow(timestamp, probe, update, op_id, io_context, stats, result);
     if (s.ok()) {
       mirrored_count++;
       #ifdef NDEBUG
       // In non-DEBUG builds, we can break as soon as we find the correct
       // rowset to mirror to. In a DEBUG build, though, we keep looking
       // through all, and make sure that we only update in one of them.
       break;
       #endif
     } else if (!s.IsNotFound()) {
       RETURN_NOT_OK_PREPEND(s, Substitute("Unable to mirror update to rowset $0 for key: $1",
           new_rowset->ToString(), probe.schema()->CreateKeyProjection().DebugRow(probe.row_key())));
     }
     // IsNotFound is OK - it might be in a different one.
   }
   CHECK_EQ(mirrored_count, 1)
     << "Updated row in compaction input, but didn't mirror in exactly 1 new rowset: "
     << probe.schema()->CreateKeyProjection().DebugRow(probe.row_key());
   return Status::OK();
 }

 Status DuplicatingRowSet::CheckRowPresent(const RowSetKeyProbe &probe, const IOContext* io_context,
                                           bool *present, ProbeStats* stats) const {
   *present = false;
   for (const shared_ptr<RowSet> &rowset : old_rowsets_) {
     RETURN_NOT_OK(rowset->CheckRowPresent(probe, io_context, present, stats));
     if (*present) {
       return Status::OK();
     }
   }
   return Status::OK();
 }

 Status DuplicatingRowSet::CountRows(const IOContext* io_context, rowid_t *count) const {
   int64_t accumulated_count = 0;
   for (const shared_ptr<RowSet> &rs : new_rowsets_) {
     rowid_t this_count;
     RETURN_NOT_OK(rs->CountRows(io_context, &this_count));
     accumulated_count += this_count;
   }

   CHECK_LT(accumulated_count, std::numeric_limits<rowid_t>::max())
     << "TODO: should make sure this is 64-bit safe - probably not right now"
     << " because rowid_t is only 32-bit.";
   *count = accumulated_count;
   return Status::OK();
 }

 Status DuplicatingRowSet::CountLiveRows(uint64_t* count) const {
   for (const shared_ptr<RowSet>& rs : old_rowsets_) {
     uint64_t tmp = 0;
     RETURN_NOT_OK(rs->CountLiveRows(&tmp));
     *count += tmp;
   }
   return Status::OK();
 }

 Status DuplicatingRowSet::GetBounds(string* min_encoded_key,
                                     string* max_encoded_key) const {
   // The range out of the output rowset always spans the full range
   // of the input rowsets, since no new rows can be inserted.
   // The output rowsets are in ascending order, so their total range
   // spans the range [front().min, back().max].
   string junk;
   RETURN_NOT_OK(new_rowsets_.front()->GetBounds(min_encoded_key, &junk));
   RETURN_NOT_OK(new_rowsets_.back()->GetBounds(&junk, max_encoded_key));
   return Status::OK();
 }

 uint64_t DuplicatingRowSet::OnDiskSize() const {
   uint64_t size = 0;
   for (const shared_ptr<RowSet> &rs : new_rowsets_) {
     size += rs->OnDiskSize();
   }
   return size;
 }

 uint64_t DuplicatingRowSet::OnDiskBaseDataSize() const {
   uint64_t size = 0;
   for (const shared_ptr<RowSet> &rs : new_rowsets_) {
     size += rs->OnDiskBaseDataSize();
   }
   return size;
 }

 uint64_t DuplicatingRowSet::OnDiskBaseDataColumnSize(const ColumnId& col_id) const {
   uint64_t size = 0;
   for (const shared_ptr<RowSet> &rs : new_rowsets_) {
     size += rs->OnDiskBaseDataColumnSize(col_id);
   }
   return size;
 }

 uint64_t DuplicatingRowSet::OnDiskBaseDataSizeWithRedos() const {
   // The actual value of this doesn't matter, since it won't be selected
   // for compaction.
   uint64_t size = 0;
   for (const shared_ptr<RowSet> &rs : new_rowsets_) {
     size += rs->OnDiskBaseDataSizeWithRedos();
   }
   return size;
 }

 shared_ptr<RowSetMetadata> DuplicatingRowSet::metadata() {
   return shared_ptr<RowSetMetadata>(reinterpret_cast<RowSetMetadata *>(NULL));
 }

 Status DuplicatingRowSet::DebugDump(vector<string> *lines) { // NOLINT(*)
   int i = 1;
   for (const shared_ptr<RowSet> &rs : old_rowsets_) {
     LOG_STRING(INFO, lines) << "Duplicating rowset input " << ToString() << " "
                             << i << "/" << old_rowsets_.size() << ":";
     RETURN_NOT_OK(rs->DebugDump(lines));
     i++;
   }
   i = 1;
   for (const shared_ptr<RowSet> &rs : new_rowsets_) {
     LOG_STRING(INFO, lines) << "Duplicating rowset output " << ToString() << " "
                             << i << "/" << new_rowsets_.size() << ":";
     RETURN_NOT_OK(rs->DebugDump(lines));
     i++;
   }

   return Status::OK();
 }

 } // namespace tablet
 } // namespace kudu
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.

	#include "kudu/tablet/rowset.h"

	#include <limits>
	#include <memory>
	#include <string>
	#include <type_traits>
	#include <utility>
	#include <vector>

	#include "kudu/common/generic_iterators.h"
	#include "kudu/common/iterator.h"
	#include "kudu/common/schema.h"
	#include "kudu/common/timestamp.h"
	#include "kudu/gutil/strings/substitute.h"
	#include "kudu/tablet/rowset_metadata.h"

	using std::shared_ptr;
	using std::string;
	using std::unique_ptr;
	using std::vector;
	using strings::Substitute;

	namespace kudu {

	using fs::IOContext;

	namespace tablet {

	RowIteratorOptions::RowIteratorOptions()
	: projection(nullptr),
	snap_to_include(MvccSnapshot::CreateSnapshotIncludingAllOps()),
	order(OrderMode::UNORDERED),
	include_deleted_rows(false) {}

	Status RowSet::NewRowIteratorWithBounds(const RowIteratorOptions& opts,
	IterWithBounds* out) const {
	// Get the iterator.
	unique_ptr<RowwiseIterator> iter;
	RETURN_NOT_OK(NewRowIterator(opts, &iter));

	// Get the bounds. Some rowsets (e.g. MRS) have no bounds; that's OK as the
	// bounds aren't required.
	string lower;
	string upper;
	Status s = GetBounds(&lower, &upper);
	if (s.ok()) {
	out->encoded_bounds = std::make_pair(std::move(lower), std::move(upper));
	} else if (!s.IsNotSupported()) {
	RETURN_NOT_OK(s);
	}
	out->iter = std::move(iter);
	return Status::OK();
	}

	DuplicatingRowSet::DuplicatingRowSet(RowSetVector old_rowsets,
	RowSetVector new_rowsets)
	: old_rowsets_(std::move(old_rowsets)),
	new_rowsets_(std::move(new_rowsets)) {
	CHECK_GT(old_rowsets_.size(), 0);
	CHECK_GT(new_rowsets_.size(), 0);
	}

	DuplicatingRowSet::~DuplicatingRowSet() {
	}

	// Stringify the given list of rowsets into 'dst'.
	static void AppendRowSetStrings(const RowSetVector &rowsets, string *dst) {
	bool first = true;
	dst->append("[");
	for (const shared_ptr<RowSet> &rs : rowsets) {
	if (!first) {
	dst->append(", ");
	}
	first = false;
	dst->append(rs->ToString());
	}
	dst->append("]");
	}

	string DuplicatingRowSet::ToString() const {
	string ret;
	ret.append("DuplicatingRowSet(");

	AppendRowSetStrings(old_rowsets_, &ret);
	ret.append(" -> ");
	AppendRowSetStrings(new_rowsets_, &ret);
	ret.append(")");
	return ret;
	}

	Status DuplicatingRowSet::NewRowIterator(const RowIteratorOptions& opts,
	unique_ptr<RowwiseIterator>* out) const {
	// Use the original rowset.
	if (old_rowsets_.size() == 1) {
	return old_rowsets_[0]->NewRowIterator(opts, out);
	}
	// Union or merge between them

	vector<IterWithBounds> iters;
	for (const auto& rs : old_rowsets_) {
	IterWithBounds iwb;
	RETURN_NOT_OK_PREPEND(rs->NewRowIteratorWithBounds(opts, &iwb),
	Substitute("Could not create iterator for rowset $0",
	rs->ToString()));
	iters.emplace_back(std::move(iwb));
	}

	switch (opts.order) {
	case ORDERED:
	*out = NewMergeIterator(MergeIteratorOptions(opts.include_deleted_rows), std::move(iters));
	break;
	case UNORDERED:
	*out = NewUnionIterator(std::move(iters));
	break;
	default:
	LOG(FATAL) << "unknown order: " << opts.order;
	}
	return Status::OK();
	}

	Status DuplicatingRowSet::NewCompactionInput(const Schema* /projection/,
	const MvccSnapshot& /snap/,
	const IOContext* /io_context/,
	unique_ptr<CompactionInput>* /out/) const {
	LOG(FATAL) << "duplicating rowsets do not act as compaction input";
	return Status::OK();
	}

	Status DuplicatingRowSet::MutateRow(Timestamp timestamp,
	const RowSetKeyProbe &probe,
	const RowChangeList &update,
	const consensus::OpId& op_id,
	const IOContext* io_context,
	ProbeStats* stats,
	OperationResultPB* result) {
	// Duplicate the update to both the relevant input rowset and the output rowset.
	//
	// It's crucial to do the mutation against the input side first, due to the potential
	// for a race during flush: the output rowset may not yet hold a DELETE which
	// is present in the input rowset. In that case, the UPDATE against the output rowset would
	// succeed whereas it can't be applied to the input rowset. So, we update the input rowset first,
	// and if it succeeds, propagate to the output.

	// First mutate the relevant input rowset.
	bool updated = false;
	for (const shared_ptr<RowSet> &rowset : old_rowsets_) {
	Status s = rowset->MutateRow(timestamp, probe, update, op_id, io_context, stats, result);
	if (s.ok()) {
	updated = true;
	break;
	} else if (!s.IsNotFound()) {
	LOG(ERROR) << "Unable to update key "
	<< probe.schema()->CreateKeyProjection().DebugRow(probe.row_key())
	<< " (failed on rowset " << rowset->ToString() << "): "
	<< s.ToString();
	return s;
	}
	}

	if (!updated) {
	return Status::NotFound("not found in any compaction input");
	}

	// If it succeeded there, we also need to mirror into the new rowset.
	int mirrored_count = 0;
	for (const shared_ptr<RowSet> &new_rowset : new_rowsets_) {
	Status s = new_rowset->MutateRow(timestamp, probe, update, op_id, io_context, stats, result);
	if (s.ok()) {
	mirrored_count++;
	#ifdef NDEBUG
	// In non-DEBUG builds, we can break as soon as we find the correct
	// rowset to mirror to. In a DEBUG build, though, we keep looking
	// through all, and make sure that we only update in one of them.
	break;
	#endif
	} else if (!s.IsNotFound()) {
	RETURN_NOT_OK_PREPEND(s, Substitute("Unable to mirror update to rowset $0 for key: $1",
	new_rowset->ToString(), probe.schema()->CreateKeyProjection().DebugRow(probe.row_key())));
	}
	// IsNotFound is OK - it might be in a different one.
	}
	CHECK_EQ(mirrored_count, 1)
	<< "Updated row in compaction input, but didn't mirror in exactly 1 new rowset: "
	<< probe.schema()->CreateKeyProjection().DebugRow(probe.row_key());
	return Status::OK();
	}

	Status DuplicatingRowSet::CheckRowPresent(const RowSetKeyProbe &probe, const IOContext* io_context,
	bool present, ProbeStats stats) const {
	*present = false;
	for (const shared_ptr<RowSet> &rowset : old_rowsets_) {
	RETURN_NOT_OK(rowset->CheckRowPresent(probe, io_context, present, stats));
	if (*present) {
	return Status::OK();
	}
	}
	return Status::OK();
	}

	Status DuplicatingRowSet::CountRows(const IOContext* io_context, rowid_t *count) const {
	int64_t accumulated_count = 0;
	for (const shared_ptr<RowSet> &rs : new_rowsets_) {
	rowid_t this_count;
	RETURN_NOT_OK(rs->CountRows(io_context, &this_count));
	accumulated_count += this_count;
	}

	CHECK_LT(accumulated_count, std::numeric_limits<rowid_t>::max())
	<< "TODO: should make sure this is 64-bit safe - probably not right now"
	<< " because rowid_t is only 32-bit.";
	*count = accumulated_count;
	return Status::OK();
	}

	Status DuplicatingRowSet::CountLiveRows(uint64_t* count) const {
	for (const shared_ptr<RowSet>& rs : old_rowsets_) {
	uint64_t tmp = 0;
	RETURN_NOT_OK(rs->CountLiveRows(&tmp));
	*count += tmp;
	}
	return Status::OK();
	}

	Status DuplicatingRowSet::GetBounds(string* min_encoded_key,
	string* max_encoded_key) const {
	// The range out of the output rowset always spans the full range
	// of the input rowsets, since no new rows can be inserted.
	// The output rowsets are in ascending order, so their total range
	// spans the range [front().min, back().max].
	string junk;
	RETURN_NOT_OK(new_rowsets_.front()->GetBounds(min_encoded_key, &junk));
	RETURN_NOT_OK(new_rowsets_.back()->GetBounds(&junk, max_encoded_key));
	return Status::OK();
	}

	uint64_t DuplicatingRowSet::OnDiskSize() const {
	uint64_t size = 0;
	for (const shared_ptr<RowSet> &rs : new_rowsets_) {
	size += rs->OnDiskSize();
	}
	return size;
	}

	uint64_t DuplicatingRowSet::OnDiskBaseDataSize() const {
	uint64_t size = 0;
	for (const shared_ptr<RowSet> &rs : new_rowsets_) {
	size += rs->OnDiskBaseDataSize();
	}
	return size;
	}

	uint64_t DuplicatingRowSet::OnDiskBaseDataColumnSize(const ColumnId& col_id) const {
	uint64_t size = 0;
	for (const shared_ptr<RowSet> &rs : new_rowsets_) {
	size += rs->OnDiskBaseDataColumnSize(col_id);
	}
	return size;
	}

	uint64_t DuplicatingRowSet::OnDiskBaseDataSizeWithRedos() const {
	// The actual value of this doesn't matter, since it won't be selected
	// for compaction.
	uint64_t size = 0;
	for (const shared_ptr<RowSet> &rs : new_rowsets_) {
	size += rs->OnDiskBaseDataSizeWithRedos();
	}
	return size;
	}

	shared_ptr<RowSetMetadata> DuplicatingRowSet::metadata() {
	return shared_ptr<RowSetMetadata>(reinterpret_cast<RowSetMetadata *>(NULL));
	}

	Status DuplicatingRowSet::DebugDump(vector<string> lines) { // NOLINT()
	int i = 1;
	for (const shared_ptr<RowSet> &rs : old_rowsets_) {
	LOG_STRING(INFO, lines) << "Duplicating rowset input " << ToString() << " "
	<< i << "/" << old_rowsets_.size() << ":";
	RETURN_NOT_OK(rs->DebugDump(lines));
	i++;
	}
	i = 1;
	for (const shared_ptr<RowSet> &rs : new_rowsets_) {
	LOG_STRING(INFO, lines) << "Duplicating rowset output " << ToString() << " "
	<< i << "/" << new_rowsets_.size() << ":";
	RETURN_NOT_OK(rs->DebugDump(lines));
	i++;
	}

	return Status::OK();
	}

	} // namespace tablet
	} // namespace kudu