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 14:36:36 UTC
[1/4] kudu git commit: KUDU-1706 Unable to build from source using
Python 3
Repository: kudu
Updated Branches:
refs/heads/master 0233ceb60 -> 92e426bf0
KUDU-1706 Unable to build from source using Python 3
Fixed python preflight and postflight scripts to support python
version 2 and 3. Converted print statements to have parenthese
print(). Replaced file() with preferred open(). Resolved some string
concatentation issues. Added universal_newlines=True) to
subprocess.Popen in shell function of postflight.py script because
in python3 it would expect bytes instead of a string if not set.
Tested on python 2.6.6, 2.7.12, 3.5.2
Change-Id: I3fcc92cb185c3dc880e06b4d235c094c8b8cd4cf
Reviewed-on: http://gerrit.cloudera.org:8080/5153
Tested-by: Kudu Jenkins
Reviewed-by: Jordan Birdsell <jt...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/fd813731
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/fd813731
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/fd813731
Branch: refs/heads/master
Commit: fd81373177dbd159861df38b25ebec73d0efd0de
Parents: 0233ceb
Author: Andy Stadtler <an...@phdata.io>
Authored: Sat Nov 19 18:32:00 2016 -0600
Committer: Jordan Birdsell <jt...@apache.org>
Committed: Mon Nov 21 03:17:31 2016 +0000
----------------------------------------------------------------------
thirdparty/postflight.py | 13 +++++++------
thirdparty/preflight.py | 41 +++++++++++++++++++++--------------------
2 files changed, 28 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/fd813731/thirdparty/postflight.py
----------------------------------------------------------------------
diff --git a/thirdparty/postflight.py b/thirdparty/postflight.py
index 0a20823..e61adcb 100755
--- a/thirdparty/postflight.py
+++ b/thirdparty/postflight.py
@@ -32,7 +32,8 @@ def shell(status_msg, script):
p = subprocess.Popen(["/bin/bash"],
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
- stdin=subprocess.PIPE)
+ stdin=subprocess.PIPE,
+ universal_newlines=True)
stdout, _ = p.communicate(input=script)
if p.returncode != 0:
to_print += " FAILED\n"
@@ -40,7 +41,7 @@ def shell(status_msg, script):
to_print += "==> " + line + "\n"
else:
to_print += " PASSED"
- print to_print
+ print(to_print)
if p.returncode != 0:
sys.exit(1)
@@ -84,11 +85,11 @@ echo All TSAN dependencies checked
def main():
- print "Running post-flight checks"
- print "-------------------------"
+ print("Running post-flight checks")
+ print("-------------------------")
check_tsan_dependencies()
- print "-------------------------"
- print "Post-flight checks succeeded."
+ print("-------------------------")
+ print("Post-flight checks succeeded.")
return 0
if __name__ == "__main__":
http://git-wip-us.apache.org/repos/asf/kudu/blob/fd813731/thirdparty/preflight.py
----------------------------------------------------------------------
diff --git a/thirdparty/preflight.py b/thirdparty/preflight.py
index 1f6ba15..88e0a83 100755
--- a/thirdparty/preflight.py
+++ b/thirdparty/preflight.py
@@ -32,7 +32,7 @@ import platform
import sys
import subprocess
-DEV_NULL = file("/dev/null", "wb")
+DEV_NULL = open("/dev/null", "wb")
CC = os.environ.get("CC", "cc")
CXX = os.environ.get("CXX", "c++")
@@ -57,17 +57,17 @@ else:
REQUIRED_TOOLS.append("libtool")
def log_failure_and_exit(error_msg, exc=None):
- print "***", error_msg
+ print("*** " + error_msg)
if exc:
- print exc.message
- print
- print "Please refer to docs/installation.adoc for complete installation "
- print "instructions for your platform."
+ print(exc.message)
+ print("")
+ print("Please refer to docs/installation.adoc for complete installation ")
+ print("instructions for your platform.")
sys.exit(1)
def try_do(status_msg, error_msg, func):
- print status_msg, "..."
+ print(status_msg + " ...")
try:
func()
except Exception as e:
@@ -81,9 +81,10 @@ def compile(script, flags=None):
p = subprocess.Popen([CXX, '-o', '/dev/null', '-c', '-x', 'c++', '-'] + flags,
stderr=subprocess.PIPE,
stdout=subprocess.PIPE,
- stdin=subprocess.PIPE)
- print >>p.stdin, script
- stdout, stderr = p.communicate()
+ stdin=subprocess.PIPE,
+ universal_newlines=True)
+
+ stdout, stderr = p.communicate(input=script)
if p.returncode != 0:
raise Exception("Compilation failed: " + stderr)
@@ -92,7 +93,7 @@ def check_tools():
""" Check that basic build tools are installed. """
missing = []
for tool in REQUIRED_TOOLS:
- print "Checking for " + tool
+ print("Checking for " + tool)
if subprocess.call(["which", tool], stdout=DEV_NULL, stderr=DEV_NULL) != 0:
missing.append(tool)
if missing:
@@ -130,18 +131,18 @@ def check_sasl():
def main():
- print "Running pre-flight checks"
- print "-------------------------"
- print "Using C compiler:", CXX
- print "Using C++ compiler:", CXX
- print ""
- print " (Set $CC and $CXX to change compiler)"
- print "-------------------------"
+ print("Running pre-flight checks")
+ print("-------------------------")
+ print("Using C compiler: " + CXX)
+ print("Using C++ compiler: " + CXX)
+ print("")
+ print(" (Set $CC and $CXX to change compiler)")
+ print("-------------------------")
check_tools()
check_cxx11()
check_sasl()
- print "-------------"
- print "Pre-flight checks succeeded."
+ print("-------------")
+ print("Pre-flight checks succeeded.")
return 0
if __name__ == "__main__":
[4/4] kudu git commit: rpc-test: fix flakiness on keepalive test
Posted by to...@apache.org.
rpc-test: fix flakiness on keepalive test
The keepalive test was using a pretty short keepalive and asserting on
the observed timing. In addition, the test was setting the keepalive
scanner and timer to a granularity equal to the keepalive time, which
meant that connections could time out much more quickly than expected,
eg:
- t=1000: timer tick runs, sets reactor's cur_time_ to t=1000ms
- t=1149: connection starts, sets last_activity_time_ = 1000ms
- t=1150: timer tick runs, sets reactor's cur_time to 1050ms
scanner sees that the connection has been "idle" for 50ms
To fix this, the patch sets the timer granularity to 20% of the
keepalive time, to ensure that at most we are 20% off from the intended
time instead of 100% off.
This reduced the keepalive test flakiness from ~10% to 0%. One other
test case still appears to be flaky, but will address that separately.
Change-Id: I0900ad45e07c936186e3477dd0b82cb226e20361
Reviewed-on: http://gerrit.cloudera.org:8080/5159
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/92e426bf
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/92e426bf
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/92e426bf
Branch: refs/heads/master
Commit: 92e426bf04fb154ec55108ee4e1df2d3463acdac
Parents: 949d195
Author: Todd Lipcon <to...@apache.org>
Authored: Sun Nov 20 19:44:29 2016 -0800
Committer: Todd Lipcon <to...@apache.org>
Committed: Mon Nov 21 14:35:42 2016 +0000
----------------------------------------------------------------------
src/kudu/rpc/rpc-test-base.h | 5 ++++-
src/kudu/rpc/rpc-test.cc | 4 ++--
2 files changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/92e426bf/src/kudu/rpc/rpc-test-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc-test-base.h b/src/kudu/rpc/rpc-test-base.h
index 1e2d4fc..133cc14 100644
--- a/src/kudu/rpc/rpc-test-base.h
+++ b/src/kudu/rpc/rpc-test-base.h
@@ -427,8 +427,11 @@ class RpcTestBase : public KuduTest {
bld.set_num_reactors(n_reactors);
bld.set_connection_keepalive_time(
MonoDelta::FromMilliseconds(keepalive_time_ms_));
+ // In order for the keepalive timing to be accurate, we need to scan connections
+ // significantly more frequently than the keepalive time. This "coarse timer"
+ // granularity determines this.
bld.set_coarse_timer_granularity(MonoDelta::FromMilliseconds(
- std::min(keepalive_time_ms_, 100)));
+ std::min(keepalive_time_ms_ / 5, 100)));
bld.set_metric_entity(metric_entity_);
std::shared_ptr<Messenger> messenger;
CHECK_OK(bld.Build(&messenger));
http://git-wip-us.apache.org/repos/asf/kudu/blob/92e426bf/src/kudu/rpc/rpc-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc-test.cc b/src/kudu/rpc/rpc-test.cc
index 7d9840e..4b03cf3 100644
--- a/src/kudu/rpc/rpc-test.cc
+++ b/src/kudu/rpc/rpc-test.cc
@@ -218,7 +218,7 @@ TEST_P(TestRpc, TestConnectionKeepalive) {
// Only run one reactor per messenger, so we can grab the metrics from that
// one without having to check all.
n_server_reactor_threads_ = 1;
- keepalive_time_ms_ = 50;
+ keepalive_time_ms_ = 100;
// Set up server.
Sockaddr server_addr;
@@ -243,7 +243,7 @@ TEST_P(TestRpc, TestConnectionKeepalive) {
ASSERT_EQ(0, metrics.num_server_connections_) << "Client should have 0 server connections";
ASSERT_EQ(1, metrics.num_client_connections_) << "Client should have 1 client connections";
- SleepFor(MonoDelta::FromMilliseconds(100));
+ SleepFor(MonoDelta::FromMilliseconds(200));
// After sleeping, the keepalive timer should have closed both sides of
// the connection.
[2/4] kudu git commit: Make
RowChangeListDecoder::RemoveColumnIdsFromChangeList() take a
RowChangeListEncoder as an out param
Posted by to...@apache.org.
Make RowChangeListDecoder::RemoveColumnIdsFromChangeList() take a RowChangeListEncoder as an out param
All the other methods in this class, of this genre, take one but
this one was taking the raw faststring buffer instead. Though it
did have the advantage of not forcing the caller to clear() the
buffer everytime it was weirdly out of place and might introduce
subtle bugs with new semantics for RowChangeListEncoder.
Change-Id: I6584b22662b58227a80eaebb7d89199d3ea2498a
Reviewed-on: http://gerrit.cloudera.org:8080/4928
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: David Ribeiro Alves <dr...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/01b1366e
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/01b1366e
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/01b1366e
Branch: refs/heads/master
Commit: 01b1366e44766b61d9f562be004f0299d516b121
Parents: fd81373
Author: David Alves <dr...@apache.org>
Authored: Thu Nov 3 03:16:32 2016 -0700
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Mon Nov 21 05:43:51 2016 +0000
----------------------------------------------------------------------
src/kudu/common/row_changelist.cc | 12 +++++-------
src/kudu/common/row_changelist.h | 7 +++----
src/kudu/tablet/memrowset.cc | 7 +++++--
3 files changed, 13 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/01b1366e/src/kudu/common/row_changelist.cc
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_changelist.cc b/src/kudu/common/row_changelist.cc
index d5f69cc..2ca7ac4 100644
--- a/src/kudu/common/row_changelist.cc
+++ b/src/kudu/common/row_changelist.cc
@@ -168,15 +168,13 @@ Status RowChangeListDecoder::Init() {
Status RowChangeListDecoder::ProjectChangeList(const DeltaProjector& projector,
const RowChangeList& src,
- faststring* buf) {
+ RowChangeListEncoder* out) {
RowChangeListDecoder decoder(src);
RETURN_NOT_OK(decoder.Init());
-
- buf->clear();
- RowChangeListEncoder encoder(buf);
+ CHECK_EQ(DCHECK_NOTNULL(out)->type_, RowChangeList::kUninitialized);
if (decoder.is_delete()) {
- encoder.SetToDelete();
+ out->SetToDelete();
return Status::OK();
}
@@ -193,8 +191,8 @@ Status RowChangeListDecoder::ProjectChangeList(const DeltaProjector& projector,
continue;
}
- encoder.SetType(decoder.type_);
- encoder.EncodeColumnMutationRaw(dec.col_id, dec.null, dec.raw_value);
+ 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/01b1366e/src/kudu/common/row_changelist.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_changelist.h b/src/kudu/common/row_changelist.h
index 57566b7..ed5f536 100644
--- a/src/kudu/common/row_changelist.h
+++ b/src/kudu/common/row_changelist.h
@@ -355,12 +355,11 @@ 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 and will be empty if the projection
- // ends up producing an empty RowChangeList.
+ // The projected RowChangeList will be encoded in 'out'.
+ // Requires that 'out' is non-null and that it's uninitialized.
static Status ProjectChangeList(const DeltaProjector& projector,
const RowChangeList& src,
- faststring* buf);
+ RowChangeListEncoder* out);
// 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
http://git-wip-us.apache.org/repos/asf/kudu/blob/01b1366e/src/kudu/tablet/memrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/memrowset.cc b/src/kudu/tablet/memrowset.cc
index c787164..04cc100 100644
--- a/src/kudu/tablet/memrowset.cc
+++ b/src/kudu/tablet/memrowset.cc
@@ -599,12 +599,15 @@ Status MemRowSet::Iterator::GetCurrentRow(RowBlockRow* dst_row,
for (const Mutation *mut = src_row.redo_head();
mut != nullptr;
mut = mut->acquire_next()) {
+
+ delta_buf_.clear();
+ RowChangeListEncoder enc(&delta_buf_);
RETURN_NOT_OK(RowChangeListDecoder::ProjectChangeList(delta_projector_,
mut->changelist(),
- &delta_buf_));
+ &enc));
// The projection resulted in an empty mutation (e.g. update of a removed column)
- if (delta_buf_.size() == 0) continue;
+ if (enc.is_empty()) continue;
Mutation *mutation = Mutation::CreateInArena(mutation_arena,
mut->timestamp(),
[3/4] kudu git commit: Don't do UNDO garbage collection until after
the REDO->UNDO transformation
Posted by to...@apache.org.
Don't do UNDO garbage collection until after the REDO->UNDO transformation
Reasoning about what has been garbage collected and not while at the
same time generating UNDOs is tricky. Particularly taking REINSERTS
into account and multiple row versions merging (part of a follow up
patch).
This moves UNDO garbage collection to after the transformation and
simplifies things a bit, at the cost of some extra work.
Change-Id: I10e4bbb1cac68474756ea1716cfc4c5420cf503e
Reviewed-on: http://gerrit.cloudera.org:8080/4993
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: David Ribeiro Alves <dr...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/949d1958
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/949d1958
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/949d1958
Branch: refs/heads/master
Commit: 949d1958f8754f4d27213d5f0a19fd6c02af7fcc
Parents: 01b1366
Author: David Alves <dr...@apache.org>
Authored: Mon Nov 7 22:07:56 2016 -0800
Committer: David Ribeiro Alves <dr...@apache.org>
Committed: Mon Nov 21 05:44:08 2016 +0000
----------------------------------------------------------------------
src/kudu/tablet/compaction.cc | 126 ++++++++++++++-----------------
src/kudu/tablet/compaction.h | 29 +++----
src/kudu/tablet/delta_compaction.cc | 13 ++--
3 files changed, 72 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/949d1958/src/kudu/tablet/compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc
index 7f47753..459cf9b 100644
--- a/src/kudu/tablet/compaction.cc
+++ b/src/kudu/tablet/compaction.cc
@@ -519,6 +519,15 @@ class MergeCompactionInput : public CompactionInput {
Arena* prepared_block_arena_;
};
+// Makes the current head point to the 'new_head', if it's not null, and
+// makes 'new_head' the new head.
+void SetHead(Mutation** current_head, Mutation* new_head) {
+ if (*current_head != nullptr) {
+ new_head->set_next(*current_head);
+ }
+ *current_head = new_head;
+}
+
} // anonymous namespace
string RowToString(const RowBlockRow& row, const Mutation* redo_head, const Mutation* undo_head) {
@@ -606,23 +615,39 @@ void RowSetsInCompaction::DumpToLog() const {
}
}
-void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow* row) {
+void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts,
+ Mutation** undo_head,
+ const Mutation* redo_head,
+ bool* is_garbage_collected) {
+ *is_garbage_collected = false;
if (!history_gc_opts.gc_enabled()) {
return;
}
+ // Make sure there is at most one REDO in the redo_head and that, if present, it's a DELETE.
+ if (redo_head != nullptr) {
+ CHECK(redo_head->changelist().is_delete());
+ CHECK(redo_head->next() == nullptr);
+
+ // Garbage collect rows that are deleted before the AHM.
+ if (history_gc_opts.IsAncientHistory(redo_head->timestamp())) {
+ *is_garbage_collected = true;
+ return;
+ }
+ }
+
DVLOG(5) << "Ancient history mark: " << history_gc_opts.ancient_history_mark().ToString()
<< ": " << HybridClock::StringifyTimestamp(history_gc_opts.ancient_history_mark());
Mutation *prev_undo = nullptr;
- Mutation *undo_mut = row->undo_head;
+ Mutation *undo_mut = *undo_head;
while (undo_mut != nullptr) {
if (history_gc_opts.IsAncientHistory(undo_mut->timestamp())) {
// Drop all undos following this one in the list; Their timestamps will be lower.
if (prev_undo != nullptr) {
prev_undo->set_next(nullptr);
} else {
- row->undo_head = nullptr;
+ *undo_head = nullptr;
}
break;
}
@@ -631,30 +656,12 @@ 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,
- const Schema* base_schema,
Mutation** new_undo_head,
Mutation** new_redo_head,
Arena* arena,
- RowBlockRow* dst_row,
- bool* is_garbage_collected,
- uint64_t* num_rows_history_truncated) {
- // In most cases, we don't want to remove all traces of this row after
- // compaction.
- *is_garbage_collected = false;
+ RowBlockRow* dst_row) {
// At the time of writing, REINSERT is never encoded as an UNDO and the base
// data can never be in a deleted state (see KUDU-237).
@@ -663,7 +670,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
#define ERROR_LOG_CONTEXT \
Substitute("Source Row: $0\nDest Row: $1", \
CompactionInputRowToString(src_row), \
- RowToString(*dst_row, undo_head, redo_head))
+ RowToString(*dst_row, undo_head, redo_delete))
faststring dst;
RowChangeListEncoder undo_encoder(&dst);
@@ -672,10 +679,9 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
// which doesn't actually mutate it and having Mutation::set_next()
// take a non-const value is required in other places.
Mutation* undo_head = const_cast<Mutation*>(src_row.undo_head);
- Mutation* redo_head = nullptr;
+ Mutation* redo_delete = nullptr;
- // Convert the redos into undos. Any redos that are eligible for history GC
- // are applied and then thrown away (they don't generate undos).
+ // Convert the redos into undos.
for (const Mutation *redo_mut = src_row.redo_head;
redo_mut != nullptr;
redo_mut = redo_mut->acquire_next()) {
@@ -689,11 +695,11 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
Mutation* current_undo;
DVLOG(3) << " @" << redo_mut->timestamp() << ": "
- << redo_mut->changelist().ToString(*base_schema);
+ << redo_mut->changelist().ToString(*src_row.row.schema());
RowChangeListDecoder redo_decoder(redo_mut->changelist());
RETURN_NOT_OK_LOG(redo_decoder.Init(), ERROR,
- "Unable to decode changelist. " + ERROR_LOG_CONTEXT);
+ "Unable to decode changelist.\n" + ERROR_LOG_CONTEXT);
switch (redo_decoder.get_type()) {
case RowChangeList::kUpdate: {
@@ -702,7 +708,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
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);
+ "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.
@@ -710,37 +716,29 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
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());
- SetUndoHead(&undo_head, current_undo);
+ SetHead(&undo_head, current_undo);
break;
}
case RowChangeList::kDelete: {
+ CHECK(redo_delete == nullptr);
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());
-
- 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;
- }
+ redo_delete = Mutation::CreateInArena(arena,
+ redo_mut->timestamp(),
+ undo_encoder.as_changelist());
break;
}
case RowChangeList::kReinsert: {
+ DCHECK(is_deleted) << "Got REINSERT for a non-deleted row. " << ERROR_LOG_CONTEXT;
+ CHECK(redo_delete != nullptr) << "Got REINSERT without a redo DELETE. "
+ << ERROR_LOG_CONTEXT;
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. We apply the reinsert to the new
@@ -763,12 +761,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
// Also reset the previous redo head since it stored the delete which was nullified
// by this reinsert
- redo_head = nullptr;
-
- if ((*num_rows_history_truncated)++ == 0) {
- LOG(WARNING) << "Found REINSERT REDO truncating row history for " << ERROR_LOG_CONTEXT
- << " Note: this warning will appear only for the first truncated row";
- }
+ redo_delete = nullptr;
if (PREDICT_FALSE(VLOG_IS_ON(2))) {
VLOG(2) << "Found REINSERT REDO, cannot create UNDO for it, resetting row history "
@@ -781,7 +774,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
}
*new_undo_head = undo_head;
- *new_redo_head = redo_head;
+ *new_redo_head = redo_delete;
return Status::OK();
@@ -799,8 +792,6 @@ Status FlushCompactionInput(CompactionInput* input,
RowBlock block(out->schema(), 100, nullptr);
- uint64_t num_rows_history_truncated = 0;
-
while (input->HasMoreBlocks()) {
RETURN_NOT_OK(input->PrepareBlock(&rows));
@@ -822,24 +813,22 @@ Status FlushCompactionInput(CompactionInput* input,
Mutation* new_undos_head = nullptr;
Mutation* new_redos_head = nullptr;
- bool is_garbage_collected;
- uint64_t prev_num_rows_history_truncated = num_rows_history_truncated;
- RemoveAncientUndos(history_gc_opts, input_row);
RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(snap,
*input_row,
- history_gc_opts,
- schema,
&new_undos_head,
&new_redos_head,
input->PreparedBlockArena(),
- &dst_row,
- &is_garbage_collected,
- &num_rows_history_truncated));
+ &dst_row));
+
+ // Remove ancient UNDOS and check whether the row should be garbage collected.
+ bool is_garbage_collected;
+ RemoveAncientUndos(history_gc_opts,
+ &new_undos_head,
+ new_redos_head,
+ &is_garbage_collected);
DVLOG(4) << "Output Row: " << RowToString(dst_row, new_redos_head, new_undos_head) <<
- "; Was garbage collected? " << is_garbage_collected <<
- "; Was history truncated? " <<
- (num_rows_history_truncated > prev_num_rows_history_truncated);
+ "; Was garbage collected? " << is_garbage_collected;
// Whether this row was garbage collected
if (is_garbage_collected) {
@@ -874,11 +863,6 @@ Status FlushCompactionInput(CompactionInput* input,
RETURN_NOT_OK(input->FinishBlock());
}
-
- if (num_rows_history_truncated > 0) {
- LOG(WARNING) << "Total " << num_rows_history_truncated
- << " rows lost some history due to REINSERT after DELETE";
- }
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/949d1958/src/kudu/tablet/compaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.h b/src/kudu/tablet/compaction.h
index a83707b..a430688 100644
--- a/src/kudu/tablet/compaction.h
+++ b/src/kudu/tablet/compaction.h
@@ -167,7 +167,16 @@ struct CompactionInputRow {
// Function shared by flushes and compactions. Removes UNDO Mutations
// considered "ancient" from the given CompactionInputRow, modifying the undo
// mutation list in-place.
-void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow* row);
+// 'is_garbage_collected': Set to true if the row was marked as deleted prior
+// to the ancient history mark, with no reinsertions after that. In such a
+// case, all traces of the row should be removed from disk by the caller.
+//
+// This is supposed to be called after ApplyMutationsAndGenerateUndos() where REDOS
+// are transformed in UNDOs. There can be at most one REDO in 'redo_head', a DELETE.
+void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts,
+ Mutation** undo_head,
+ const Mutation* redo_head,
+ bool* is_garbage_collected);
// Function shared by flushes, compactions and major delta compactions. Applies all the REDO
// mutations from 'src_row' to the 'dst_row', and generates the related UNDO mutations. Some
@@ -177,28 +186,12 @@ void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow
// - Major delta compaction: Applies only the REDOs that have corresponding columns in the schema
// belonging to 'dst_row'. Those that don't belong to that schema are
// ignored.
-//
-// 'history_gc_opts': Options indicating whether row history should be
-// garbage-collected.
-//
-// 'is_garbage_collected': Set to true if the row was marked as deleted prior
-// to the ancient history mark, with no reinsertions after that. In such a
-// case, all traces of the row should be removed from disk by the caller.
-//
-// 'num_rows_history_truncated': Set to the number of rows that had a "reinsert
-// after delete" that was compacted, resulting in the loss of history prior to
-// the reinsert. This is related to a known issue that will eventually be
-// fixed. See KUDU-237.
Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
const CompactionInputRow& src_row,
- const HistoryGcOpts& history_gc_opts,
- const Schema* base_schema,
Mutation** new_undo_head,
Mutation** new_redo_head,
Arena* arena,
- RowBlockRow* dst_row,
- bool* is_garbage_collected,
- uint64_t* num_rows_history_truncated);
+ RowBlockRow* dst_row);
// Iterate through this compaction input, flushing all rows to the given RollingDiskRowSetWriter.
// The 'snap' argument should match the MvccSnapshot used to create the compaction input.
http://git-wip-us.apache.org/repos/asf/kudu/blob/949d1958/src/kudu/tablet/delta_compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_compaction.cc b/src/kudu/tablet/delta_compaction.cc
index e25adf0..f66e402 100644
--- a/src/kudu/tablet/delta_compaction.cc
+++ b/src/kudu/tablet/delta_compaction.cc
@@ -113,7 +113,6 @@ Status MajorDeltaCompaction::FlushRowSetAndDeltas() {
DVLOG(1) << "Applying deltas and rewriting columns (" << partial_schema_.ToString() << ")";
DeltaStats redo_stats;
DeltaStats undo_stats;
- uint64_t num_rows_history_truncated = 0;
size_t nrows = 0;
// We know that we're reading everything from disk so we're including all transactions.
MvccSnapshot snap = MvccSnapshot::CreateSnapshotIncludingAllTransactions();
@@ -157,17 +156,17 @@ Status MajorDeltaCompaction::FlushRowSetAndDeltas() {
// NOTE: This is presently ignored.
bool is_garbage_collected;
- RemoveAncientUndos(history_gc_opts_, input_row);
RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(snap,
*input_row,
- history_gc_opts_,
- &base_schema_,
&new_undos_head,
&new_redos_head,
&arena,
- &dst_row,
- &is_garbage_collected,
- &num_rows_history_truncated));
+ &dst_row));
+
+ RemoveAncientUndos(history_gc_opts_,
+ &new_undos_head,
+ new_redos_head,
+ &is_garbage_collected);
DVLOG(3) << "MDC Output Row - RowId: " << row_id << " "
<< RowToString(dst_row, new_undos_head, new_redos_head);