You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/11/21 03:18:59 UTC
kudu git commit: KUDU-237 (part 1) - Support proper mutation encoding
for reinserts
Repository: kudu
Updated Branches:
refs/heads/master d8ce43893 -> 0233ceb60
KUDU-237 (part 1) - Support proper mutation encoding for reinserts
Currently reinserts are not properly encoded like the other mutations
and instead refer to an in-memory row, meaning we can't flush them
as they don't include the indirect data.
This patch makes it so that reinserts are properly encoded, like other
mutations but does not yet include the step of flushing them to disk,
which will be done in a follow-up patch.
Change-Id: I2b0a95e3f618bf20eb18ca4f4b03b9e2ce5704e8
Reviewed-on: http://gerrit.cloudera.org:8080/4791
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/0233ceb6
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/0233ceb6
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/0233ceb6
Branch: refs/heads/master
Commit: 0233ceb6011c5c7cd3c267d68d0e821dd9e8fd1c
Parents: d8ce438
Author: David Alves <dr...@apache.org>
Authored: Fri Oct 21 16:59:36 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Mon Nov 21 02:57:21 2016 +0000
----------------------------------------------------------------------
src/kudu/common/row_changelist-test.cc | 69 ++++++++----
src/kudu/common/row_changelist.cc | 161 ++++++++++++++--------------
src/kudu/common/row_changelist.h | 119 +++++++++++++-------
src/kudu/tablet/compaction.cc | 136 ++++++++++++-----------
src/kudu/tablet/memrowset-test.cc | 3 +-
src/kudu/tablet/memrowset.cc | 35 ++----
6 files changed, 290 insertions(+), 233 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/0233ceb6/src/kudu/common/row_changelist-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_changelist-test.cc b/src/kudu/common/row_changelist-test.cc
index c5cfa73..1394f9e 100644
--- a/src/kudu/common/row_changelist-test.cc
+++ b/src/kudu/common/row_changelist-test.cc
@@ -21,6 +21,7 @@
#include "kudu/common/schema.h"
#include "kudu/common/row_changelist.h"
#include "kudu/common/row.h"
+#include "kudu/common/rowblock.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/faststring.h"
#include "kudu/util/hexdump.h"
@@ -128,31 +129,56 @@ TEST_F(TestRowChangeList, TestDeletes) {
}
TEST_F(TestRowChangeList, TestReinserts) {
- RowBuilder rb(schema_);
- rb.AddString(Slice("hello"));
- rb.AddString(Slice("world"));
- rb.AddUint32(12345);
- rb.AddNull();
// Construct a REINSERT.
faststring buf;
- RowChangeListEncoder rcl(&buf);
- rcl.SetToReinsert(rb.data());
+ RowChangeListEncoder reinsert_1_enc(&buf);
+
+ {
+ // Reinserts include indirect data, so it should be ok to make the RowBuilder a scoped var.
+ RowBuilder rb(schema_);
+ rb.AddString(Slice("hello"));
+ rb.AddString(Slice("world"));
+ rb.AddUint32(12345);
+ rb.AddNull();
+ reinsert_1_enc.SetToReinsert(rb.row());
+ }
LOG(INFO) << "Encoded: " << HexDump(buf);
// Read it back.
- EXPECT_EQ(string("REINSERT (string col1=hello, string col2=world, "
- "uint32 col3=12345, uint32 col4=NULL)"),
+ // Note that col1 (hello) is not present in the output string as it's part of the primary
+ // key which not encoded in the REINSERT mutation.
+ EXPECT_EQ(string("REINSERT col2=world, col3=12345, col4=NULL"),
RowChangeList(Slice(buf)).ToString(schema_));
- RowChangeListDecoder decoder((RowChangeList(buf)));
- ASSERT_OK(decoder.Init());
- ASSERT_TRUE(decoder.is_reinsert());
+ RowChangeListDecoder reinsert_1_dec((RowChangeList(buf)));
+ ASSERT_OK(reinsert_1_dec.Init());
+ ASSERT_TRUE(reinsert_1_dec.is_reinsert());
+
+ faststring buf2;
+ RowChangeListEncoder reinsert_2_enc(&buf2);
+ {
+ RowBlock block(schema_, 1, nullptr);
+ RowBlockRow dst_row = block.row(0);
+ RowBuilder rb(schema_);
+ rb.AddString(Slice("hello"));
+ rb.AddString(Slice("mundo"));
+ rb.AddUint32(54321);
+ rb.AddUint32(1);
+ CopyRow(rb.row(), &dst_row, static_cast<Arena*>(nullptr));
+ reinsert_2_enc.SetToReinsert();
+ CHECK_OK(reinsert_1_dec.MutateRowAndCaptureChanges(&dst_row,
+ static_cast<Arena*>(nullptr),
+ &reinsert_2_enc));
+ // The row should now match reinsert 1
+ ASSERT_STR_CONTAINS(schema_.DebugRow(dst_row),
+ "(string col1=hello, string col2=world, uint32 col3=12345, uint32 col4=NULL)");
+ }
- Slice s;
- ASSERT_OK(decoder.GetReinsertedRowSlice(schema_, &s));
- ASSERT_EQ(s, rb.data());
+ // And reinsert 2 should contain the original state of the row.
+ EXPECT_EQ(string("REINSERT col2=mundo, col3=54321, col4=1"),
+ RowChangeList(Slice(buf2)).ToString(schema_));
}
TEST_F(TestRowChangeList, TestInvalid_EmptySlice) {
@@ -175,17 +201,17 @@ TEST_F(TestRowChangeList, TestInvalid_TooLongDelete) {
TEST_F(TestRowChangeList, TestInvalid_TooShortReinsert) {
RowChangeListDecoder decoder(RowChangeList(Slice("\x03")));
- ASSERT_OK(decoder.Init());
- Slice s;
- ASSERT_STR_CONTAINS(decoder.GetReinsertedRowSlice(schema_, &s).ToString(),
- "Corruption: REINSERT changelist wrong length");
+ ASSERT_STR_CONTAINS(decoder.Init().ToString(),
+ "Corruption: empty changelist - expected column updates");
}
TEST_F(TestRowChangeList, TestInvalid_SetNullForNonNullableColumn) {
faststring buf;
RowChangeListEncoder rcl(&buf);
+
// Set column 0 = NULL
- rcl.AddRawColumnUpdate(schema_.column_id(0), true, Slice());
+ rcl.SetToUpdate();
+ rcl.EncodeColumnMutationRaw(schema_.column_id(0), true, Slice());
ASSERT_EQ("[invalid update: Corruption: decoded set-to-NULL "
"for non-nullable column: col1[string NOT NULL], "
@@ -198,7 +224,8 @@ TEST_F(TestRowChangeList, TestInvalid_SetWrongSizeForIntColumn) {
RowChangeListEncoder rcl(&buf);
// Set column id 2 = \xff
// (column id 2 is UINT32, so should be 4 bytes)
- rcl.AddRawColumnUpdate(schema_.column_id(2), false, Slice("\xff"));
+ rcl.SetToUpdate();
+ rcl.EncodeColumnMutationRaw(schema_.column_id(2), false, Slice("\xff"));
ASSERT_EQ("[invalid update: Corruption: invalid value \\xff "
"for column col3[uint32 NOT NULL], "
http://git-wip-us.apache.org/repos/asf/kudu/blob/0233ceb6/src/kudu/common/row_changelist.cc
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_changelist.cc b/src/kudu/common/row_changelist.cc
index 5429c9c..d5f69cc 100644
--- a/src/kudu/common/row_changelist.cc
+++ b/src/kudu/common/row_changelist.cc
@@ -43,15 +43,16 @@ string RowChangeList::ToString(const Schema &schema) const {
if (decoder.is_delete()) {
return string("DELETE");
- } else if (decoder.is_reinsert()) {
- ConstContiguousRow row(&schema, decoder.remaining_);
- return string("REINSERT ") + schema.DebugRow(row);
+ }
+
+ string ret;
+ if (decoder.is_reinsert()) {
+ ret = "REINSERT ";
} else {
CHECK(decoder.is_update()) << "Unknown changelist type!";
+ ret = "SET ";
}
- string ret = "SET ";
-
bool first = true;
while (decoder.HasNext()) {
if (!first) {
@@ -95,26 +96,19 @@ string RowChangeList::ToString(const Schema &schema) const {
return ret;
}
-void RowChangeListEncoder::AddRawColumnUpdate(
- int col_id, bool is_null, Slice new_val) {
- if (type_ == RowChangeList::kUninitialized) {
- SetType(RowChangeList::kUpdate);
- } else {
- DCHECK_EQ(RowChangeList::kUpdate, type_);
- }
-
- InlinePutVarint32(dst_, col_id);
- if (is_null) {
- dst_->push_back(0);
- } else {
- InlinePutVarint32(dst_, new_val.size() + 1);
- dst_->append(new_val.data(), new_val.size());
- }
-}
-
void RowChangeListEncoder::AddColumnUpdate(const ColumnSchema& col_schema,
int col_id,
const void* cell_ptr) {
+ SetToUpdate();
+ EncodeColumnMutation(col_schema, col_id, cell_ptr);
+}
+
+void RowChangeListEncoder::EncodeColumnMutation(const ColumnSchema& col_schema,
+ int col_id,
+ const void* cell_ptr) {
+ DCHECK_NE(RowChangeList::kUninitialized, type_);
+ DCHECK(type_ == RowChangeList::kUpdate || type_ == RowChangeList::kReinsert);
+
Slice val_slice;
if (cell_ptr != nullptr) {
if (col_schema.type_info()->physical_type() == BINARY) {
@@ -128,7 +122,22 @@ void RowChangeListEncoder::AddColumnUpdate(const ColumnSchema& col_schema,
DCHECK(col_schema.is_nullable());
}
- AddRawColumnUpdate(col_id, cell_ptr == nullptr, val_slice);
+ EncodeColumnMutationRaw(col_id, cell_ptr == nullptr, val_slice);
+}
+
+
+void RowChangeListEncoder::EncodeColumnMutationRaw(int col_id, bool is_null, Slice new_val) {
+ // The type must have been set beforehand.
+ DCHECK_NE(RowChangeList::kUninitialized, type_);
+ DCHECK(type_ == RowChangeList::kUpdate || type_ == RowChangeList::kReinsert);
+
+ InlinePutVarint32(dst_, col_id);
+ if (is_null) {
+ dst_->push_back(0);
+ } else {
+ InlinePutVarint32(dst_, new_val.size() + 1);
+ dst_->append(new_val.data(), new_val.size());
+ }
}
Status RowChangeListDecoder::Init() {
@@ -148,66 +157,56 @@ Status RowChangeListDecoder::Init() {
}
remaining_.remove_prefix(1);
- return Status::OK();
-}
-
-Status RowChangeListDecoder::GetReinsertedRowSlice(const Schema& schema, Slice* s) const {
- DCHECK(is_reinsert());
- int expected_size = ContiguousRowHelper::row_size(schema);
- if (remaining_.size() != expected_size) {
- return Status::Corruption(Substitute("REINSERT changelist wrong length (expected $0)",
- expected_size,
- remaining_.ToDebugString()));
+ // We should discard empty REINSERT/UPDATE RowChangeLists, so if after getting
+ // the type remaining_ is empty() return an error.
+ if (!is_delete() && remaining_.empty()) {
+ return Status::Corruption("empty changelist - expected column updates");
}
- *s = remaining_;
return Status::OK();
}
-Status RowChangeListDecoder::ProjectUpdate(const DeltaProjector& projector,
- const RowChangeList& src,
- faststring *buf) {
+Status RowChangeListDecoder::ProjectChangeList(const DeltaProjector& projector,
+ const RowChangeList& src,
+ faststring* buf) {
RowChangeListDecoder decoder(src);
RETURN_NOT_OK(decoder.Init());
buf->clear();
RowChangeListEncoder encoder(buf);
+
if (decoder.is_delete()) {
encoder.SetToDelete();
- } else if (decoder.is_reinsert()) {
- Slice reinsert;
- RETURN_NOT_OK(decoder.GetReinsertedRowSlice(*projector.delta_schema(), &reinsert));
-
- // ReInsert = MemStore Insert -> Delete -> (Re)Insert
- ConstContiguousRow src_row = ConstContiguousRow(projector.delta_schema(),
- reinsert);
- RowProjector row_projector(projector.delta_schema(), projector.projection());
- size_t row_size = ContiguousRowHelper::row_size(*projector.projection());
- uint8_t buffer[row_size];
- ContiguousRow row(projector.projection(), buffer);
- RETURN_NOT_OK(row_projector.Init());
- RETURN_NOT_OK(row_projector.ProjectRowForRead(src_row, &row, static_cast<Arena*>(NULL)));
- encoder.SetToReinsert(Slice(buffer, row_size));
- } else if (decoder.is_update()) {
- while (decoder.HasNext()) {
- DecodedUpdate dec;
- RETURN_NOT_OK(decoder.DecodeNext(&dec));
- int col_idx;
- const void* new_val;
- RETURN_NOT_OK(dec.Validate(*projector.projection(), &col_idx, &new_val));
- // If the new schema doesn't have this column, throw away the update.
- if (col_idx == Schema::kColumnNotFound) {
- continue;
- }
+ return Status::OK();
+ }
- encoder.AddRawColumnUpdate(dec.col_id, dec.null, dec.raw_value);
+ DCHECK(decoder.is_reinsert() || decoder.is_update());
+
+ while (decoder.HasNext()) {
+ DecodedUpdate dec;
+ RETURN_NOT_OK(decoder.DecodeNext(&dec));
+ int col_idx;
+ const void* new_val;
+ RETURN_NOT_OK(dec.Validate(*projector.projection(), &col_idx, &new_val));
+ // If the new schema doesn't have this column, throw away the update.
+ if (col_idx == Schema::kColumnNotFound) {
+ continue;
}
+
+ encoder.SetType(decoder.type_);
+ encoder.EncodeColumnMutationRaw(dec.col_id, dec.null, dec.raw_value);
}
+
return Status::OK();
}
-Status RowChangeListDecoder::ApplyRowUpdate(RowBlockRow *dst_row,
- Arena *arena, RowChangeListEncoder* undo_encoder) {
+Status RowChangeListDecoder::MutateRowAndCaptureChanges(RowBlockRow* dst_row,
+ Arena* arena,
+ RowChangeListEncoder* out) {
+
+ DCHECK(is_reinsert() || is_update());
+ DCHECK(out->is_initialized());
+
const Schema* dst_schema = dst_row->schema();
while (HasNext()) {
@@ -216,6 +215,8 @@ Status RowChangeListDecoder::ApplyRowUpdate(RowBlockRow *dst_row,
int col_idx;
const void* value;
RETURN_NOT_OK(dec.Validate(*dst_schema, &col_idx, &value));
+ // Reinserts don't update keys so they shouldn't include the key columns.
+ DCHECK(!dst_schema->is_key_column(col_idx));
// If the delta is for a column ID not part of the projection
// we're scanning, just skip over it.
@@ -225,10 +226,10 @@ Status RowChangeListDecoder::ApplyRowUpdate(RowBlockRow *dst_row,
const ColumnSchema& col_schema = dst_schema->column(col_idx);
SimpleConstCell src(&col_schema, value);
-
- // save the old cell on the undo encoder
RowBlockRow::Cell dst_cell = dst_row->cell(col_idx);
- undo_encoder->AddColumnUpdate(col_schema, dec.col_id, dst_cell.ptr());
+
+ // save the old cell in 'out'
+ out->EncodeColumnMutation(col_schema, dec.col_id, dst_cell.ptr());
// copy the new cell to the row
RETURN_NOT_OK(CopyCell(src, &dst_cell, arena));
@@ -241,7 +242,7 @@ Status RowChangeListDecoder::ApplyRowUpdate(RowBlockRow *dst_row,
Status RowChangeListDecoder::ApplyToOneColumn(size_t row_idx, ColumnBlock* dst_col,
const Schema& dst_schema,
int col_idx, Arena *arena) {
- DCHECK_EQ(RowChangeList::kUpdate, type_);
+ DCHECK(is_reinsert() || is_update());
const ColumnSchema& col_schema = dst_schema.column(col_idx);
ColumnId col_id = dst_schema.column_id(col_idx);
@@ -272,19 +273,23 @@ Status RowChangeListDecoder::RemoveColumnIdsFromChangeList(const RowChangeList&
RowChangeListEncoder* out) {
RowChangeListDecoder decoder(src);
RETURN_NOT_OK(decoder.Init());
+
if (decoder.is_delete()) {
out->SetToDelete();
- } else if (decoder.is_reinsert()) {
- out->SetToReinsert(decoder.remaining_);
- } else if (decoder.is_update()) {
- while (decoder.HasNext()) {
- DecodedUpdate dec;
- RETURN_NOT_OK(decoder.DecodeNext(&dec));
- if (!std::binary_search(col_ids.begin(), col_ids.end(), dec.col_id)) {
- out->AddRawColumnUpdate(dec.col_id, dec.null, dec.raw_value);
- }
+ return Status::OK();
+ }
+
+ DCHECK(decoder.is_reinsert() || decoder.is_update());
+
+ while (decoder.HasNext()) {
+ DecodedUpdate dec;
+ RETURN_NOT_OK(decoder.DecodeNext(&dec));
+ if (!std::binary_search(col_ids.begin(), col_ids.end(), dec.col_id)) {
+ out->SetType(decoder.type_);
+ out->EncodeColumnMutationRaw(dec.col_id, dec.null, dec.raw_value);
}
}
+
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/0233ceb6/src/kudu/common/row_changelist.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_changelist.h b/src/kudu/common/row_changelist.h
index d922356..57566b7 100644
--- a/src/kudu/common/row_changelist.h
+++ b/src/kudu/common/row_changelist.h
@@ -152,16 +152,22 @@ class RowChangeListEncoder {
SetType(RowChangeList::kDelete);
}
- // TODO: This doesn't currently copy the indirected data, so
- // REINSERT deltas can't possibly work anywhere but in memory.
- // For now, there is an assertion in the DeltaFile flush code
- // that prevents us from accidentally depending on this anywhere
- // but in-memory.
- void SetToReinsert(const Slice &row_data) {
+
+ // Marks the type of the row change as an update but doesn't
+ // actually add a mutation.
+ void SetToUpdate() {
+ SetType(RowChangeList::kUpdate);
+ }
+
+ void SetToReinsert() {
SetType(RowChangeList::kReinsert);
- dst_->append(row_data.data(), row_data.size());
}
+ // Encodes a REINSERT for 'src_row'.
+ // Both direct and indirect data from 'src_row' will be copied and encoded.
+ template<class RowType>
+ void SetToReinsert(const RowType& src_row);
+
// Add a column update, given knowledge of the schema.
//
// If 'cell_ptr' is NULL, then 'col_schema' must refer to a nullable
@@ -184,8 +190,13 @@ class RowChangeListEncoder {
return type_ != RowChangeList::kUninitialized;
}
+ // Returns whether this encoder has an empty RowChangeList.
+ // Returns true if no type was set, or if the type was set to REINSERT or UPDATE but no
+ // actual changes were encoded.
bool is_empty() {
- return dst_->size() == 0;
+ if (dst_->size() == 0) return true;
+ return dst_->size() == 1 &&
+ (type_ == RowChangeList::kReinsert || type_ == RowChangeList::kUpdate);
}
private:
@@ -193,13 +204,26 @@ class RowChangeListEncoder {
FRIEND_TEST(TestRowChangeList, TestInvalid_SetWrongSizeForIntColumn);
friend class RowChangeListDecoder;
+ // Sets the type of the changelist to 'type' if it's not set.
+ // If a type *is* set then, in debug, this checks that the new
+ // type is the same as the current type.
void SetType(RowChangeList::ChangeType type) {
- DCHECK_EQ(type_, RowChangeList::kUninitialized);
- type_ = type;
- dst_->push_back(type);
+ if (!is_initialized()) {
+ DCHECK(is_empty());
+ type_ = type;
+ dst_->push_back(type);
+ return;
+ }
+ DCHECK_EQ(type_, type);
}
- // Add a column update by a raw value. This allows copying RCLs
+ // Internal version of AddColumnUpdate which does not set type, allowing
+ // it to work for both REINSERT and UPDATE.
+ void EncodeColumnMutation(const ColumnSchema& col_schema,
+ int col_id,
+ const void* cell_ptr);
+
+ // Add a column mutation by a raw value. This allows copying RCLs
// from one file to another without having any awareness of schema.
//
// If 'is_null' is set, then encodes a SET [col_id]=NULL update.
@@ -208,13 +232,32 @@ class RowChangeListEncoder {
// 'new_val' is the encoded form of the new value. In the case of
// a STRING, this is the actual user-provided STRING. Otherwise,
// it is the fixed-length representation of the type.
- void AddRawColumnUpdate(int col_id, bool is_null, Slice new_val);
-
+ void EncodeColumnMutationRaw(int col_id, bool is_null, Slice new_val);
RowChangeList::ChangeType type_;
faststring *dst_;
};
+template<class RowType>
+void RowChangeListEncoder::SetToReinsert(const RowType& src_row) {
+ DCHECK_EQ(RowChangeList::kUninitialized, type_);
+ SetType(RowChangeList::kReinsert);
+ const Schema* schema = src_row.schema();
+ for (int i = 0; i < schema->num_columns(); ++i) {
+ // Reinserts don't need to store the keys.
+ if (schema->is_key_column(i)) continue;
+ ColumnId col_id = schema->column_id(i);
+ const ColumnSchema& col_schema = schema->column(i);
+ if (col_schema.is_nullable() && src_row.is_null(i)) {
+ EncodeColumnMutationRaw(col_id, true /* null */, Slice());
+ continue;
+ }
+
+ const void* cell_ptr = DCHECK_NOTNULL(src_row.cell_ptr(i));
+ EncodeColumnMutation(col_schema, col_id, cell_ptr);
+ }
+}
+
class RowChangeListDecoder {
public:
@@ -259,10 +302,6 @@ class RowChangeListDecoder {
return type_ == RowChangeList::kReinsert;
}
- // If this RCL is a REINSERT, then returns the reinserted row in
- // the contiguous in-memory row format.
- Status GetReinsertedRowSlice(const Schema& schema, Slice* s) const;
-
const RowChangeList::ChangeType get_type() const {
return type_;
}
@@ -282,15 +321,21 @@ class RowChangeListDecoder {
}
// Applies changes in this decoder to the specified row and saves the old
- // state of the row into the undo_encoder.
- Status ApplyRowUpdate(RowBlockRow *dst_row,
- Arena *arena, RowChangeListEncoder* undo_encoder);
+ // state of the row into 'out'. Does not set a type in 'out', which must be pre-set.
+ //
+ // If the Schema of 'dst_row' does not include any of the columns in the encoded RowChangeList
+ // out->is_empty() will return true.
+ //
+ // REQUIRES: is_update() or is_reinsert()
+ Status MutateRowAndCaptureChanges(RowBlockRow* dst_row,
+ Arena* arena,
+ RowChangeListEncoder* out);
- // Apply this UPDATE RowChangeList to row number 'row_idx' in 'dst_col', but only
- // any updates that correspond to column 'col_idx' of 'dst_schema'.
+ // Apply this UPDATE or REINSERT RowChangeList to row number 'row_idx' in 'dst_col',
+ // but only any updates that correspond to column 'col_idx' of 'dst_schema'.
// Any indirect data is copied into 'arena' if non-NULL.
//
- // REQUIRES: is_update()
+ // REQUIRES: is_update() or is_reinsert()
Status ApplyToOneColumn(size_t row_idx, ColumnBlock* dst_col,
const Schema& dst_schema, int col_idx, Arena *arena);
@@ -311,19 +356,18 @@ class RowChangeListDecoder {
// Project the 'src' RowChangeList using the delta 'projector'
// The projected RowChangeList will be encoded to specified 'buf'.
- // The buffer will be cleared before adding the result.
- static Status ProjectUpdate(const DeltaProjector& projector,
- const RowChangeList& src,
- faststring *buf);
-
- // If 'src' is an update, then only add changes for columns NOT
- // specified by 'column_indexes' to 'out'. Delete and Re-insert
- // changes are added to 'out' as-is. If an update only contained
- // changes for 'column_indexes', then out->is_initialized() will
+ // The buffer will be cleared before adding the result and will be empty if the projection
+ // ends up producing an empty RowChangeList.
+ static Status ProjectChangeList(const DeltaProjector& projector,
+ const RowChangeList& src,
+ faststring* buf);
+
+ // If 'src' is an UPDATE or REINSERT, then only add changes for columns NOT specified by
+ // 'column_indexes' to 'out'. Delete changes are added to 'out' as-is. If an update only
+ // contained changes for 'column_indexes', then out->is_initialized() will
// return false.
- // 'column_ids' must be sorted; 'out' must be
- // valid for the duration of this method, but not have been
- // previously initialized.
+ // 'column_ids' must be sorted; 'out' must be valid for the duration of this method, but not
+ // have been previously initialized.
static Status RemoveColumnIdsFromChangeList(const RowChangeList& src,
const std::vector<ColumnId>& column_ids,
RowChangeListEncoder* out);
@@ -367,12 +411,13 @@ class RowChangeListDecoder {
// The update->raw_value slice points to memory within the buffer
// being decoded by this object. No copies are made.
//
- // REQUIRES: is_update()
+ // REQUIRES: is_update() or is_reinsert()
Status DecodeNext(DecodedUpdate* update);
private:
FRIEND_TEST(TestRowChangeList, TestEncodeDecodeUpdates);
friend class RowChangeList;
+ friend class RowChangeListEncoder;
// Data remaining in the source buffer.
// This slice is advanced forward as entries are decoded.
http://git-wip-us.apache.org/repos/asf/kudu/blob/0233ceb6/src/kudu/tablet/compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc
index addea71..7f47753 100644
--- a/src/kudu/tablet/compaction.cc
+++ b/src/kudu/tablet/compaction.cc
@@ -631,6 +631,17 @@ void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow
}
}
+namespace {
+// Makes the current UNDO head point to the 'new_head', if it's not null, and
+// makes 'new_head' the new UNDO head.
+void SetUndoHead(Mutation** current_head, Mutation* new_head) {
+ if (*current_head != nullptr) {
+ new_head->set_next(*current_head);
+ }
+ *current_head = new_head;
+}
+} // anonymous namespace
+
Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
const CompactionInputRow& src_row,
const HistoryGcOpts& history_gc_opts,
@@ -645,8 +656,6 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
// compaction.
*is_garbage_collected = false;
- const Schema* dst_schema = dst_row->schema();
-
// At the time of writing, REINSERT is never encoded as an UNDO and the base
// data can never be in a deleted state (see KUDU-237).
bool is_deleted = false;
@@ -683,67 +692,70 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
<< redo_mut->changelist().ToString(*base_schema);
RowChangeListDecoder redo_decoder(redo_mut->changelist());
- Status s = redo_decoder.Init();
- if (PREDICT_FALSE(!s.ok())) {
- LOG(ERROR) << "Unable to decode changelist. " << ERROR_LOG_CONTEXT;
- return s;
- }
+ RETURN_NOT_OK_LOG(redo_decoder.Init(), ERROR,
+ "Unable to decode changelist. " + ERROR_LOG_CONTEXT);
- if (redo_decoder.is_update()) {
- DCHECK(!is_deleted) << "Got UPDATE for deleted row. " << ERROR_LOG_CONTEXT;
+ switch (redo_decoder.get_type()) {
+ case RowChangeList::kUpdate: {
+ DCHECK(!is_deleted) << "Got UPDATE for deleted row. " << ERROR_LOG_CONTEXT;
- s = redo_decoder.ApplyRowUpdate(dst_row,
- static_cast<Arena *>(nullptr), &undo_encoder);
- if (PREDICT_FALSE(!s.ok())) {
- LOG(ERROR) << "Unable to apply update/create undo: " << s.ToString()
- << "\n" << ERROR_LOG_CONTEXT;
- return s;
- }
+ undo_encoder.SetToUpdate();
+ RETURN_NOT_OK_LOG(redo_decoder.MutateRowAndCaptureChanges(
+ dst_row, static_cast<Arena*>(nullptr), &undo_encoder), ERROR,
+ "Unable to apply update undo. \n " + ERROR_LOG_CONTEXT);
- // If all of the updates were for columns that we aren't projecting, we don't
- // need to push them into the UNDO file.
- if (undo_encoder.is_empty()) {
- continue;
- }
+ // If all of the updates were for columns that we aren't projecting, we don't
+ // need to push them into the UNDO file.
+ if (undo_encoder.is_empty()) {
+ continue;
+ }
- if (history_gc_opts.IsAncientHistory(redo_mut->timestamp())) {
- // Don't generate an undo for this.
- continue;
- }
+ if (history_gc_opts.IsAncientHistory(redo_mut->timestamp())) {
+ // Don't generate an undo for this.
+ continue;
+ }
- // create the UNDO mutation in the provided arena.
- current_undo = Mutation::CreateInArena(arena, redo_mut->timestamp(),
- undo_encoder.as_changelist());
+ // create the UNDO mutation in the provided arena.
+ current_undo = Mutation::CreateInArena(arena, redo_mut->timestamp(),
+ undo_encoder.as_changelist());
- // In the case where the previous undo was a nullptr just make this one
- // the head.
- if (undo_head == nullptr) {
- undo_head = current_undo;
- } else {
- current_undo->set_next(undo_head);
- undo_head = current_undo;
+ SetUndoHead(&undo_head, current_undo);
+ break;
}
+ case RowChangeList::kDelete: {
+ redo_decoder.TwiddleDeleteStatus(&is_deleted);
+ // Delete mutations are left as redos. Encode the DELETE as a redo.
+ undo_encoder.SetToDelete();
+ redo_head = Mutation::CreateInArena(arena,
+ redo_mut->timestamp(),
+ undo_encoder.as_changelist());
-
- } else if (redo_decoder.is_delete() || redo_decoder.is_reinsert()) {
- redo_decoder.TwiddleDeleteStatus(&is_deleted);
-
- if (redo_decoder.is_reinsert()) {
+ if (history_gc_opts.IsAncientHistory(redo_mut->timestamp())) {
+ // This row was deleted prior to the ancient history mark. Remove all
+ // traces of this row. This flag is set to false once again if there
+ // is a subsequent reinsert (see above).
+ *is_garbage_collected = true;
+ }
+ break;
+ }
+ case RowChangeList::kReinsert: {
+ redo_decoder.TwiddleDeleteStatus(&is_deleted);
*is_garbage_collected = false;
// Right now when a REINSERT mutation is found it is treated as a new insert and it
- // clears the whole row history before it.
-
- // Copy the reinserted row over.
- Slice reinserted_slice;
- RETURN_NOT_OK(redo_decoder.GetReinsertedRowSlice(*dst_schema, &reinserted_slice));
- ConstContiguousRow reinserted(dst_schema, reinserted_slice.data());
- // No need to copy into an arena -- can refer to the mutation's arena.
- Arena* null_arena = nullptr;
- RETURN_NOT_OK(CopyRow(reinserted, dst_row, null_arena));
-
- // Create an undo for the REINSERT
+ // clears the whole row history before it. We apply the reinsert to the new
+ // row but disregard the undo reinsert that is created.
+ //
+ // TODO(dralves) KUDU-237 Actually store the REINSERT
+ undo_encoder.SetToReinsert();
+ RETURN_NOT_OK_LOG(redo_decoder.MutateRowAndCaptureChanges(
+ dst_row, static_cast<Arena*>(nullptr), &undo_encoder), ERROR,
+ "Unable to apply reinsert undo. \n " + ERROR_LOG_CONTEXT);
+
+ // Discard the REINSERT and store a DELETE instead.
+ undo_encoder.Reset();
undo_encoder.SetToDelete();
+
// Reset the UNDO head, losing all previous undos.
undo_head = Mutation::CreateInArena(arena,
redo_mut->timestamp(),
@@ -760,27 +772,11 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
if (PREDICT_FALSE(VLOG_IS_ON(2))) {
VLOG(2) << "Found REINSERT REDO, cannot create UNDO for it, resetting row history "
- "under snapshot: " << snap.ToString() << ERROR_LOG_CONTEXT;
- }
-
- // Is a delete.
- } else {
- // Delete mutations are left as redos. Encode the DELETE as a redo.
- undo_encoder.SetToDelete();
- redo_head = Mutation::CreateInArena(arena,
- redo_mut->timestamp(),
- undo_encoder.as_changelist());
-
- if (history_gc_opts.IsAncientHistory(redo_mut->timestamp())) {
- // This row was deleted prior to the ancient history mark. Remove all
- // traces of this row. This flag is set to false once again if there
- // is a subsequent reinsert (see above).
- *is_garbage_collected = true;
+ "under snapshot: " << snap.ToString() << ERROR_LOG_CONTEXT;
}
-
+ break;
}
- } else {
- LOG(FATAL) << "Unknown mutation type!" << ERROR_LOG_CONTEXT;
+ default: LOG(FATAL) << "Unknown mutation type!" << ERROR_LOG_CONTEXT;
}
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/0233ceb6/src/kudu/tablet/memrowset-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset-test.cc b/src/kudu/tablet/memrowset-test.cc
index 021de32..02c83eb 100644
--- a/src/kudu/tablet/memrowset-test.cc
+++ b/src/kudu/tablet/memrowset-test.cc
@@ -405,8 +405,7 @@ TEST_F(TestMemRowSet, TestDelete) {
ASSERT_OK(mrs->DebugDump(&rows));
ASSERT_EQ(1, rows.size());
EXPECT_EQ("@1: row (string key=hello world, uint32 val=1) mutations="
- "[@2(DELETE), "
- "@5(REINSERT (string key=hello world, uint32 val=2))]",
+ "[@2(DELETE), @5(REINSERT val=2)]",
rows[0]);
// Verify that iterating the rowset at the first snapshot shows the row.
http://git-wip-us.apache.org/repos/asf/kudu/blob/0233ceb6/src/kudu/tablet/memrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc
index 11c4e0c..c787164 100644
--- a/src/kudu/tablet/memrowset.cc
+++ b/src/kudu/tablet/memrowset.cc
@@ -189,19 +189,10 @@ Status MemRowSet::Insert(Timestamp timestamp,
Status MemRowSet::Reinsert(Timestamp timestamp, const ConstContiguousRow& row, MRSRow *ms_row) {
DCHECK_SCHEMA_EQ(schema_, *row.schema());
- // TODO(perf): This path makes some unnecessary copies that could be reduced,
- // but let's assume that REINSERT is really rare and code for clarity over speed
- // here.
-
- // Make a copy of the row, and relocate any of its indirected data into
- // our Arena.
- DEFINE_MRSROW_ON_STACK(this, row_copy, row_copy_slice);
- RETURN_NOT_OK(row_copy.CopyRow(row, arena_.get()));
-
- // Encode the REINSERT mutation from the relocated row copy.
+ // Encode the REINSERT mutation
faststring buf;
RowChangeListEncoder encoder(&buf);
- encoder.SetToReinsert(row_copy.row_slice());
+ encoder.SetToReinsert(row);
// Move the REINSERT mutation itself into our Arena.
Mutation *mut = Mutation::CreateInArena(arena_.get(), timestamp, encoder.as_changelist());
@@ -556,19 +547,13 @@ Status MemRowSet::Iterator::ApplyMutationsToProjectedRow(
RETURN_NOT_OK(decoder.Init());
if (decoder.is_delete()) {
decoder.TwiddleDeleteStatus(&is_deleted);
- } else if (decoder.is_reinsert()) {
- decoder.TwiddleDeleteStatus(&is_deleted);
-
- Slice reinserted_slice;
- RETURN_NOT_OK(decoder.GetReinsertedRowSlice(memrowset_->schema_nonvirtual(),
- &reinserted_slice));
- ConstContiguousRow reinserted(&memrowset_->schema_nonvirtual(),
- reinserted_slice);
- RETURN_NOT_OK(projector_->ProjectRowForRead(reinserted, dst_row, dst_arena));
} else {
- DCHECK(decoder.is_update());
+ DCHECK(decoder.is_update() || decoder.is_reinsert());
+ if (decoder.is_reinsert()) {
+ decoder.TwiddleDeleteStatus(&is_deleted);
+ }
- // TODO: this is slow, since it makes multiple passes through the rowchangelist.
+ // TODO(todd): this is slow, since it makes multiple passes through the rowchangelist.
// Instead, we should keep the backwards mapping of columns.
for (const RowProjector::ProjectionIdxMapping& mapping : projector_->base_cols_mapping()) {
RowChangeListDecoder decoder(mut->changelist());
@@ -614,9 +599,9 @@ Status MemRowSet::Iterator::GetCurrentRow(RowBlockRow* dst_row,
for (const Mutation *mut = src_row.redo_head();
mut != nullptr;
mut = mut->acquire_next()) {
- RETURN_NOT_OK(RowChangeListDecoder::ProjectUpdate(delta_projector_,
- mut->changelist(),
- &delta_buf_));
+ RETURN_NOT_OK(RowChangeListDecoder::ProjectChangeList(delta_projector_,
+ mut->changelist(),
+ &delta_buf_));
// The projection resulted in an empty mutation (e.g. update of a removed column)
if (delta_buf_.size() == 0) continue;