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/09/09 19:13:43 UTC
[1/4] kudu git commit: Tie log retention to consensus watermarks
Repository: kudu
Updated Branches:
refs/heads/master 30a8337f0 -> be719edc3
Tie log retention to consensus watermarks
This changes the calculation of log retention to consult consensus.
Consensus returns a struct which indicates the watermark necessary for
durability (the committed index) as well as the watermark necessary to
catch up other peers. The durability watermark is then further
constrained by the LogAnchorRegistry, as before, to ensure that no entry
corresponding to yet-unflushed data can be GCed.
This new struct containing the "for-durability" and "for-peers"
watermarks replaces the old single "must be retained" watermark that the
log GC code used before.
The new struct is passed down into the log, and we use the following
policy:
- we always maintain any logs necessary for durability (the minimum of
those entries needed by consensus, and those entries needed by the
tablet itself)
- beyond that, we try to retain logs to catch up lagging peers, however
we never maintain more than --log_max_segments_to_retain (a new
configuration)
I removed the old flag --log_min_seconds_to_retain, since its main
purpose was for dealing with lagging peers, and that is now handled by
directly consulting consensus.
The one tricky bit of the policy is that, even though the peer catch-up
figures into log retention, we do _not_ want it to impact the
calculation of flush priority. In other words, even if the user is OK
retaining 10GB of logs to catch up trailing peers, they probably still
want to flush more aggressively than that so they can avoid very long
startup times. So, the peer-based watermark is not used during the
mapping of log anchors to retention amounts.
Note that the above is only relevant once we have implemented KUDU-38:
we currently will replay all of the retained logs even though we are
aggressively flushing to keep the durability-related retention bounded.
In practice, even without KUDU-38, this patch shouldn't have a large
negative effect on restart times. In fact, in many cases it can
_improve_ startup times, because in most steady workloads we don't have
peers that are extremely far behind. Our log retention only increases in
those cases, and only on those tablets which have a lagging follower.
For other tablets, the new retention policies actually serve to reduce
the number of retained segments, so if there are no laggy peers, we'll
start up faster.
Manually tested for now as follows:
- started a three-node cluster (locally), set to roll logs at 1MB
segments, but otherwise default
- started an insert workload against a single-tablet table
- I could see that the three servers were maintaining 2 WAL segments in
their WAL directory.
- I kill -STOPped a random server while continuing to insert. I saw that
the WALs in this tablet server's directory froze as is (obviously),
and the other two kept rolling. However, because of this change, the
other servers started retaining wals starting from the point where I
had stopped the follower.
- If I let the insert workload continue, the live servers kept rolling
up until they had 10 segments (default --log_max_segments_to_retain)
at which point they dropped the oldest log.
- I verified that, during this period while the extra segments were
retained, the servers continued to flush frequently so that their
recovery time would be bounded.
- I also verified that, if I un-paused the follower before the others
had evicted it, it was able to catch up, at which point the other
servers GCed those extra logs they had been retaining.
The above scenario is also tested through modifications to
RaftConsensusITest.TestCatchupAfterOpsEvicted and various log-test test
cases.
Change-Id: Icfc071270510f3dc3c65f88d615e93c6ffb26b12
Reviewed-on: http://gerrit.cloudera.org:8080/4177
Reviewed-by: Mike Percy <mp...@apache.org>
Tested-by: Kudu Jenkins
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/e21523f5
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/e21523f5
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/e21523f5
Branch: refs/heads/master
Commit: e21523f5fac41ae272886c89a6a45049173aa94b
Parents: 30a8337
Author: Todd Lipcon <to...@apache.org>
Authored: Thu Aug 25 22:50:39 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Fri Sep 9 18:20:46 2016 +0000
----------------------------------------------------------------------
src/kudu/consensus/consensus.h | 10 ++
src/kudu/consensus/consensus.proto | 8 ++
src/kudu/consensus/consensus_peers-test.cc | 6 +-
src/kudu/consensus/consensus_queue-test.cc | 47 +++----
src/kudu/consensus/consensus_queue.cc | 18 +--
src/kudu/consensus/consensus_queue.h | 14 +-
src/kudu/consensus/log-test-base.h | 4 -
src/kudu/consensus/log-test.cc | 93 ++++---------
src/kudu/consensus/log.cc | 131 ++++++++++---------
src/kudu/consensus/log.h | 43 ++++--
src/kudu/consensus/log_reader.cc | 49 -------
src/kudu/consensus/log_reader.h | 21 ---
src/kudu/consensus/raft_consensus-test.cc | 2 +
src/kudu/consensus/raft_consensus.cc | 14 +-
src/kudu/consensus/raft_consensus.h | 2 +
.../consensus/raft_consensus_quorum-test.cc | 2 +
src/kudu/integration-tests/delete_table-test.cc | 2 +-
.../external_mini_cluster_fs_inspector.cc | 22 ++--
.../external_mini_cluster_fs_inspector.h | 17 ++-
.../integration-tests/raft_consensus-itest.cc | 55 ++++++--
src/kudu/tablet/tablet_peer-test.cc | 60 +++------
src/kudu/tablet/tablet_peer.cc | 37 +++---
src/kudu/tablet/tablet_peer.h | 4 +-
23 files changed, 326 insertions(+), 335 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/consensus.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus.h b/src/kudu/consensus/consensus.h
index 109d55f..259a1c3 100644
--- a/src/kudu/consensus/consensus.h
+++ b/src/kudu/consensus/consensus.h
@@ -36,6 +36,7 @@ class MonoDelta;
namespace log {
class Log;
+struct RetentionIndexes;
}
namespace master {
@@ -260,6 +261,15 @@ class Consensus : public RefCountedThreadSafe<Consensus> {
return Status::NotFound("Not implemented.");
}
+
+ // Return the log indexes which the consensus implementation would like to retain.
+ //
+ // The returned 'for_durability' index ensures that no logs are GCed before
+ // the operation is fully committed. The returned 'for_peers' index indicates
+ // the index of the farthest-behind peer so that the log will try to avoid
+ // GCing these before the peer has caught up.
+ virtual log::RetentionIndexes GetRetentionIndexes() = 0;
+
protected:
friend class RefCountedThreadSafe<Consensus>;
friend class tablet::TabletPeer;
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/consensus.proto
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus.proto b/src/kudu/consensus/consensus.proto
index bde2b2a..3841aee 100644
--- a/src/kudu/consensus/consensus.proto
+++ b/src/kudu/consensus/consensus.proto
@@ -335,6 +335,14 @@ message ConsensusRequestPB {
// these operations are already committed, in which case they will be
// committed during the same request.
repeated ReplicateMsg ops = 6;
+
+ // The lowest index that is known to be replicated by all members of
+ // the configuration.
+ //
+ // NOTE: this is not necessarily monotonically increasing. For example, if a node is in
+ // the process of being added to the configuration but has not yet copied a snapshot,
+ // this value may drop to 0.
+ optional int64 all_replicated_index = 9;
}
message ConsensusResponsePB {
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/consensus_peers-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_peers-test.cc b/src/kudu/consensus/consensus_peers-test.cc
index 7159867..44927e5 100644
--- a/src/kudu/consensus/consensus_peers-test.cc
+++ b/src/kudu/consensus/consensus_peers-test.cc
@@ -113,7 +113,7 @@ class ConsensusPeersTest : public KuduTest {
// This must be called _before_ the operation is committed.
void WaitForCommitIndex(int index) {
AssertEventually([&]() {
- ASSERT_GE(message_queue_->GetCommittedIndexForTests(), index);
+ ASSERT_GE(message_queue_->GetCommittedIndex(), index);
});
}
@@ -203,7 +203,7 @@ TEST_F(ConsensusPeersTest, TestRemotePeers) {
// Wait until all peers have replicated the message, otherwise
// when we add the next one remote_peer2 might find the next message
// in the queue and will replicate it, which is not what we want.
- while (message_queue_->GetAllReplicatedIndexForTests() != first.index()) {
+ while (message_queue_->GetAllReplicatedIndex() != first.index()) {
SleepFor(MonoDelta::FromMilliseconds(1));
}
@@ -213,7 +213,7 @@ TEST_F(ConsensusPeersTest, TestRemotePeers) {
// We should not see it committed, even after 10ms,
// since only the local peer replicates the message.
SleepFor(MonoDelta::FromMilliseconds(10));
- ASSERT_LT(message_queue_->GetCommittedIndexForTests(), 2);
+ ASSERT_LT(message_queue_->GetCommittedIndex(), 2);
// Signal one of the two remote peers.
remote_peer1->SignalRequest();
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/consensus_queue-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_queue-test.cc b/src/kudu/consensus/consensus_queue-test.cc
index 33a5f6c..ff5051b 100644
--- a/src/kudu/consensus/consensus_queue-test.cc
+++ b/src/kudu/consensus/consensus_queue-test.cc
@@ -245,6 +245,7 @@ TEST_F(ConsensusQueueTest, TestGetPagedMessages) {
ConsensusRequestPB page_size_estimator;
page_size_estimator.set_caller_term(14);
page_size_estimator.set_committed_index(0);
+ page_size_estimator.set_all_replicated_index(0);
page_size_estimator.mutable_preceding_id()->CopyFrom(MinimumOpId());
// We're going to add 100 messages to the queue so we make each page fetch 9 of those,
@@ -311,7 +312,7 @@ TEST_F(ConsensusQueueTest, TestPeersDontAckBeyondWatermarks) {
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), 0);
// Since we're tracking a single peer still this should have moved the all
// replicated watermark to the last op appended to the local log.
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 100);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 100);
// Start to track the peer after the queue has some messages in it
// at a point that is halfway through the current messages in the queue.
@@ -326,7 +327,7 @@ TEST_F(ConsensusQueueTest, TestPeersDontAckBeyondWatermarks) {
ASSERT_TRUE(more_pending);
// Tracking a peer a new peer should have moved the all replicated watermark back.
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 0);
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), 0);
vector<ReplicateRefPtr> refs;
@@ -344,7 +345,7 @@ TEST_F(ConsensusQueueTest, TestPeersDontAckBeyondWatermarks) {
ASSERT_TRUE(more_pending) << "Queue didn't have anymore requests pending";
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), 100);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 100);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 100);
// if we ask for a new request, it should come back with the rest of the messages
ASSERT_OK(queue_->RequestForPeer(kPeerUuid, &request, &refs, &needs_tablet_copy));
@@ -361,7 +362,7 @@ TEST_F(ConsensusQueueTest, TestPeersDontAckBeyondWatermarks) {
WaitForLocalPeerToAckIndex(expected.index());
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), expected.index());
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), expected.index());
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), expected.index());
// extract the ops from the request to avoid double free
request.mutable_ops()->ExtractSubrange(0, request.ops_size(), nullptr);
@@ -383,7 +384,7 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
// Since only the local log has ACKed at this point,
// the committed_index should be MinimumOpId().
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 0);
// NOTE: We don't need to get operations from the queue. The queue
// only cares about what the peer reported as received, not what was sent.
@@ -401,7 +402,7 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
ASSERT_TRUE(more_pending);
// Committed index should be the same
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 0);
// Ack the first five operations for peer-2.
response.set_responder_uuid("peer-2");
@@ -412,10 +413,10 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), 5);
// However, this leader has appended operations in term 1, so we can't
// advance the committed index yet.
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 0);
// Moreover, 'peer-3' and 'peer-4' have not acked yet, so the "all-replicated"
// index also cannot advance.
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 0);
// Ack all operations for peer-3.
response.set_responder_uuid("peer-3");
@@ -429,8 +430,8 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
// Watermarks should remain the same as above: we still have not majority-replicated
// anything in the current term, so committed index cannot advance.
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), 5);
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 0);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 0);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 0);
// Ack the remaining operations for peer-4.
response.set_responder_uuid("peer-4");
@@ -440,8 +441,8 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
// Now that a majority of peers have replicated an operation in the queue's
// term the committed index should advance.
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), 10);
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 10);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 5);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 10);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 5);
}
// In this test we append a sequence of operations to a log
@@ -587,7 +588,7 @@ TEST_F(ConsensusQueueTest, TestQueueHandlesOperationOverwriting) {
// We're waiting for a two nodes. The all committed watermark should be
// 0.0 since we haven't had a successful exchange with the 'remote' peer.
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 0);
// Test even when a correct peer responds (meaning we actually get to execute
// watermark advancement) we sill have the same all-replicated watermark.
@@ -595,7 +596,7 @@ TEST_F(ConsensusQueueTest, TestQueueHandlesOperationOverwriting) {
ASSERT_OK(queue_->AppendOperation(make_scoped_refptr(new RefCountedReplicate(replicate))));
WaitForLocalPeerToAckIndex(21);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 0);
// Generate another request for the remote peer, which should include
// all of the ops since the peer's last-known committed index.
@@ -610,7 +611,7 @@ TEST_F(ConsensusQueueTest, TestQueueHandlesOperationOverwriting) {
queue_->ResponseFromPeer(response.responder_uuid(), response, &more_pending);
// Now the watermark should have advanced.
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 21);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 21);
// The messages still belong to the queue so we have to release them.
request.mutable_ops()->ExtractSubrange(0, request.ops().size(), nullptr);
@@ -649,7 +650,7 @@ TEST_F(ConsensusQueueTest, TestQueueMovesWatermarksBackward) {
// The replication watermark on a follower should not advance by virtue of appending
// entries to the log.
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), 0);
}
// Tests that we're advancing the watermarks properly and only when the peer
@@ -695,7 +696,7 @@ TEST_F(ConsensusQueueTest, TestOnlyAdvancesWatermarkWhenPeerHasAPrefixOfOurLog)
int64_t expected_all_replicated = 0;
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), expected_majority_replicated);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), expected_all_replicated);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), expected_all_replicated);
UpdatePeerWatermarkToOp(&request, &response, MakeOpId(75, 49), MinimumOpId(), 31, &more_pending);
ASSERT_TRUE(more_pending);
@@ -733,7 +734,7 @@ TEST_F(ConsensusQueueTest, TestOnlyAdvancesWatermarkWhenPeerHasAPrefixOfOurLog)
expected_majority_replicated = expected_all_replicated = 40;
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), expected_majority_replicated);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), expected_all_replicated);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), expected_all_replicated);
// Another request for this peer should get another page of messages. Still not
// on the queue's term (and thus without advancing watermarks).
@@ -751,7 +752,7 @@ TEST_F(ConsensusQueueTest, TestOnlyAdvancesWatermarkWhenPeerHasAPrefixOfOurLog)
expected_majority_replicated = expected_all_replicated = 49;
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), expected_majority_replicated);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), expected_all_replicated);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), expected_all_replicated);
// The last page of request should overwrite the peer's operations and the
// response should finally advance the watermarks.
@@ -768,7 +769,7 @@ TEST_F(ConsensusQueueTest, TestOnlyAdvancesWatermarkWhenPeerHasAPrefixOfOurLog)
queue_->ResponseFromPeer(response.responder_uuid(), response, &more_pending);
ASSERT_EQ(queue_->GetMajorityReplicatedIndexForTests(), expected_majority_replicated);
- ASSERT_EQ(queue_->GetAllReplicatedIndexForTests(), expected_all_replicated);
+ ASSERT_EQ(queue_->GetAllReplicatedIndex(), expected_all_replicated);
request.mutable_ops()->ExtractSubrange(0, request.ops().size(), nullptr);
}
@@ -823,12 +824,12 @@ TEST_F(ConsensusQueueTest, TestFollowerCommittedIndexAndMetrics) {
// The committed_index should be MinimumOpId() since UpdateFollowerCommittedIndex
// has not been called.
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 0);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 0);
// Update the committed index. In real life, this would be done by the consensus
// implementation when it receives an updated committed index from the leader.
- queue_->UpdateFollowerCommittedIndex(10);
- ASSERT_EQ(queue_->GetCommittedIndexForTests(), 10);
+ queue_->UpdateFollowerWatermarks(10, 10);
+ ASSERT_EQ(queue_->GetCommittedIndex(), 10);
// Check the metrics have the right values based on the updated committed index.
ASSERT_EQ(queue_->metrics_.num_majority_done_ops->value(), 0);
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/consensus_queue.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_queue.cc b/src/kudu/consensus/consensus_queue.cc
index dd905b8..228d723 100644
--- a/src/kudu/consensus/consensus_queue.cc
+++ b/src/kudu/consensus/consensus_queue.cc
@@ -315,6 +315,7 @@ Status PeerMessageQueue::RequestForPeer(const string& uuid,
// log entry preceding the first one in 'messages' if messages are found for the peer.
preceding_id = queue_state_.last_appended;
request->set_committed_index(queue_state_.committed_index);
+ request->set_all_replicated_index(queue_state_.all_replicated_index);
request->set_caller_term(queue_state_.current_term);
}
@@ -520,12 +521,13 @@ void PeerMessageQueue::AdvanceQueueWatermark(const char* type,
}
}
-void PeerMessageQueue::UpdateFollowerCommittedIndex(int64_t committed_index) {
- if (queue_state_.mode == NON_LEADER) {
- std::lock_guard<simple_spinlock> l(queue_lock_);
- queue_state_.committed_index = committed_index;
- UpdateMetrics();
- }
+void PeerMessageQueue::UpdateFollowerWatermarks(int64_t committed_index,
+ int64_t all_replicated_index) {
+ std::lock_guard<simple_spinlock> l(queue_lock_);
+ DCHECK_EQ(queue_state_.mode, NON_LEADER);
+ queue_state_.committed_index = committed_index;
+ queue_state_.all_replicated_index = all_replicated_index;
+ UpdateMetrics();
}
void PeerMessageQueue::NotifyPeerIsResponsiveDespiteError(const std::string& peer_uuid) {
@@ -752,12 +754,12 @@ PeerMessageQueue::TrackedPeer PeerMessageQueue::GetTrackedPeerForTests(string uu
return *tracked;
}
-int64_t PeerMessageQueue::GetAllReplicatedIndexForTests() const {
+int64_t PeerMessageQueue::GetAllReplicatedIndex() const {
std::lock_guard<simple_spinlock> lock(queue_lock_);
return queue_state_.all_replicated_index;
}
-int64_t PeerMessageQueue::GetCommittedIndexForTests() const {
+int64_t PeerMessageQueue::GetCommittedIndex() const {
std::lock_guard<simple_spinlock> lock(queue_lock_);
return queue_state_.committed_index;
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/consensus_queue.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_queue.h b/src/kudu/consensus/consensus_queue.h
index 59a570e..7b39a1a 100644
--- a/src/kudu/consensus/consensus_queue.h
+++ b/src/kudu/consensus/consensus_queue.h
@@ -219,9 +219,11 @@ class PeerMessageQueue {
const ConsensusResponsePB& response,
bool* more_pending);
- // Called by the consensus implementation to update the follower queue's
- // committed index, which is mostly used for metrics.
- void UpdateFollowerCommittedIndex(int64_t committed_index);
+ // Called by the consensus implementation to update the queue's watermarks
+ // based on information provided by the leader. This is used for metrics and
+ // log retention.
+ void UpdateFollowerWatermarks(int64_t committed_index,
+ int64_t all_replicated_index);
// Closes the queue, peers are still allowed to call UntrackPeer() and
// ResponseFromPeer() but no additional peers can be tracked or messages
@@ -230,12 +232,12 @@ class PeerMessageQueue {
virtual int64_t GetQueuedOperationsSizeBytesForTests() const;
- // Returns the last message replicated by all peers, for tests.
- virtual int64_t GetAllReplicatedIndexForTests() const;
+ // Returns the last message replicated by all peers.
+ virtual int64_t GetAllReplicatedIndex() const;
// Returns the committed index. All operations with index less than or equal to
// this index have been committed.
- virtual int64_t GetCommittedIndexForTests() const;
+ virtual int64_t GetCommittedIndex() const;
// Returns the current majority replicated index, for tests.
virtual int64_t GetMajorityReplicatedIndexForTests() const;
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/log-test-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log-test-base.h b/src/kudu/consensus/log-test-base.h
index 8c0b0b2..ca19bbd 100644
--- a/src/kudu/consensus/log-test-base.h
+++ b/src/kudu/consensus/log-test-base.h
@@ -49,8 +49,6 @@
METRIC_DECLARE_entity(tablet);
-DECLARE_int32(log_min_seconds_to_retain);
-
namespace kudu {
namespace log {
@@ -142,8 +140,6 @@ class LogTestBase : public KuduTest {
clock_.reset(new server::HybridClock());
ASSERT_OK(clock_->Init());
-
- FLAGS_log_min_seconds_to_retain = 0;
}
virtual void TearDown() OVERRIDE {
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/log-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log-test.cc b/src/kudu/consensus/log-test.cc
index b05d1e1..2042c7d 100644
--- a/src/kudu/consensus/log-test.cc
+++ b/src/kudu/consensus/log-test.cc
@@ -19,6 +19,7 @@
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <glog/stl_logging.h>
+#include <limits>
#include <vector>
#include "kudu/consensus/consensus-test-util.h"
@@ -34,6 +35,7 @@ DEFINE_int32(num_batches, 10000,
"Number of batches to write to/read from the Log in TestWriteManyBatches");
DECLARE_int32(log_min_segments_to_retain);
+DECLARE_int32(log_max_segments_to_retain);
DECLARE_double(log_inject_io_error_on_preallocate_fraction);
DECLARE_int64(fs_wal_dir_reserved_bytes);
DECLARE_int64(disk_reserved_bytes_free_for_testing);
@@ -474,7 +476,6 @@ TEST_F(LogTest, TestGCWithLogRunning) {
const int kNumOpsPerSegment = 5;
int num_gced_segments;
OpId op_id = MakeOpId(1, 1);
- int64_t anchored_index = -1;
ASSERT_OK(AppendMultiSegmentSequence(kNumTotalSegments, kNumOpsPerSegment,
&op_id, &anchors));
@@ -485,29 +486,31 @@ TEST_F(LogTest, TestGCWithLogRunning) {
// Anchors should prevent GC.
ASSERT_OK(log_->reader()->GetSegmentsSnapshot(&segments))
ASSERT_EQ(4, segments.size()) << DumpSegmentsToString(segments);
- ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&anchored_index));
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ RetentionIndexes retention;
+ ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&retention.for_durability));
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
+ ASSERT_EQ(0, num_gced_segments);
ASSERT_OK(log_->reader()->GetSegmentsSnapshot(&segments))
ASSERT_EQ(4, segments.size()) << DumpSegmentsToString(segments);
// Freeing the first 2 anchors should allow GC of them.
ASSERT_OK(log_anchor_registry_->Unregister(anchors[0]));
ASSERT_OK(log_anchor_registry_->Unregister(anchors[1]));
- ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&anchored_index));
+ ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&retention.for_durability));
// We should now be anchored on op 0.11, i.e. on the 3rd segment
- ASSERT_EQ(anchors[2]->log_index, anchored_index);
+ ASSERT_EQ(anchors[2]->log_index, retention.for_durability);
// However, first, we'll try bumping the min retention threshold and
// verify that we don't GC any.
{
google::FlagSaver saver;
FLAGS_log_min_segments_to_retain = 10;
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
ASSERT_EQ(0, num_gced_segments);
}
// Try again without the modified flag.
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
ASSERT_EQ(2, num_gced_segments) << DumpSegmentsToString(segments);
ASSERT_OK(log_->reader()->GetSegmentsSnapshot(&segments))
ASSERT_EQ(2, segments.size()) << DumpSegmentsToString(segments);
@@ -515,8 +518,8 @@ TEST_F(LogTest, TestGCWithLogRunning) {
// Release the remaining "rolled segment" anchor. GC will not delete the
// last rolled segment.
ASSERT_OK(log_anchor_registry_->Unregister(anchors[2]));
- ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&anchored_index));
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&retention.for_durability));
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
ASSERT_EQ(0, num_gced_segments) << DumpSegmentsToString(segments);
ASSERT_OK(log_->reader()->GetSegmentsSnapshot(&segments))
ASSERT_EQ(2, segments.size()) << DumpSegmentsToString(segments);
@@ -561,7 +564,7 @@ TEST_F(LogTest, TestGCOfIndexChunks) {
// Run a GC on an op in the second index chunk. We should remove only the
// earliest segment, because we are set to retain 4.
int num_gced_segments = 0;
- ASSERT_OK(log_->GC(1000006, &num_gced_segments));
+ ASSERT_OK(log_->GC(RetentionIndexes(1000006, 1000006), &num_gced_segments));
ASSERT_EQ(1, num_gced_segments);
// And we should still be able to read ops in the retained segment, even though
@@ -573,7 +576,7 @@ TEST_F(LogTest, TestGCOfIndexChunks) {
// If we drop the retention count down to 1, we can now GC, and the log index
// chunk should also be GCed.
FLAGS_log_min_segments_to_retain = 1;
- ASSERT_OK(log_->GC(1000003, &num_gced_segments));
+ ASSERT_OK(log_->GC(RetentionIndexes(1000003, 1000003), &num_gced_segments));
ASSERT_EQ(1, num_gced_segments);
Status s = log_->reader()->LookupOpId(999995, &loaded_op);
@@ -619,15 +622,14 @@ TEST_F(LogTest, TestLogReopenAndGC) {
const int kNumOpsPerSegment = 5;
int num_gced_segments;
OpId op_id = MakeOpId(1, 1);
- int64_t anchored_index = -1;
-
ASSERT_OK(AppendMultiSegmentSequence(kNumTotalSegments, kNumOpsPerSegment,
&op_id, &anchors));
// Anchors should prevent GC.
ASSERT_OK(log_->reader()->GetSegmentsSnapshot(&segments))
ASSERT_EQ(3, segments.size());
- ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&anchored_index));
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ RetentionIndexes retention;
+ ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&retention.for_durability));
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
ASSERT_OK(log_->reader()->GetSegmentsSnapshot(&segments))
ASSERT_EQ(3, segments.size());
@@ -650,18 +652,20 @@ TEST_F(LogTest, TestLogReopenAndGC) {
for (int i = 0; i < 3; i++) {
ASSERT_OK(log_anchor_registry_->Unregister(anchors[i]));
}
- ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&anchored_index));
+ ASSERT_OK(log_anchor_registry_->GetEarliestRegisteredLogIndex(&retention.for_durability));
- // If we set the min_seconds_to_retain high, then we'll retain the logs even
- // though we could GC them based on our anchoring.
- FLAGS_log_min_seconds_to_retain = 500;
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ // If we set the 'for_peers' index to indicate that these log
+ // segments are needed for catchup, that will prevent GC,
+ // even though they're no longer necessary for durability.
+ retention.for_peers = 0;
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
ASSERT_EQ(0, num_gced_segments);
+ CheckRightNumberOfSegmentFiles(4);
- // Turn off the time-based retention and try GCing again. This time
- // we should succeed.
- FLAGS_log_min_seconds_to_retain = 0;
- ASSERT_OK(log_->GC(anchored_index, &num_gced_segments));
+ // Set the max segments to retain so that, even though we have peers who need
+ // the segments, we'll GC them.
+ FLAGS_log_max_segments_to_retain = 2;
+ ASSERT_OK(log_->GC(retention, &num_gced_segments));
ASSERT_EQ(2, num_gced_segments);
// After GC there should be only one left, besides the one currently being
@@ -731,35 +735,6 @@ TEST_F(LogTest, TestLogReader) {
op.set_term(0);
SegmentSequence segments;
- // Queries for segment prefixes (used for GC)
-
- // Asking the reader the prefix of segments that does not include op 1
- // should return the empty set.
- ASSERT_OK(reader.GetSegmentPrefixNotIncluding(1, &segments));
- ASSERT_TRUE(segments.empty());
-
- // .. same for op 10
- ASSERT_OK(reader.GetSegmentPrefixNotIncluding(10, &segments));
- ASSERT_TRUE(segments.empty());
-
- // Asking for the prefix of segments not including op 20 should return
- // the first segment, since 20 is the first operation in segment 3.
- ASSERT_OK(reader.GetSegmentPrefixNotIncluding(20, &segments));
- ASSERT_EQ(segments.size(), 1);
- ASSERT_EQ(segments[0]->header().sequence_number(), 2);
-
- // Asking for 30 should include the first two.
- ASSERT_OK(reader.GetSegmentPrefixNotIncluding(30, &segments));
- ASSERT_EQ(segments.size(), 2);
- ASSERT_EQ(segments[0]->header().sequence_number(), 2);
- ASSERT_EQ(segments[1]->header().sequence_number(), 3);
-
- // Asking for anything higher should return all segments.
- ASSERT_OK(reader.GetSegmentPrefixNotIncluding(1000, &segments));
- ASSERT_EQ(segments.size(), 3);
- ASSERT_EQ(segments[0]->header().sequence_number(), 2);
- ASSERT_EQ(segments[1]->header().sequence_number(), 3);
-
// Queries for specific segment sequence numbers.
scoped_refptr<ReadableLogSegment> segment = reader.GetSegmentBySequenceNumber(2);
ASSERT_EQ(2, segment->header().sequence_number());
@@ -993,7 +968,7 @@ TEST_F(LogTest, TestReadLogWithReplacedReplicates) {
}
int num_gced = 0;
- ASSERT_OK(log_->GC(gc_index, &num_gced));
+ ASSERT_OK(log_->GC(RetentionIndexes(gc_index), &num_gced));
gc_index += rng.Uniform(10);
}
}
@@ -1016,24 +991,20 @@ TEST_F(LogTest, TestGetMaxIndexesToSegmentSizeMap) {
// Check getting all the segments we can get rid of (5 - 2).
log_->GetMaxIndexesToSegmentSizeMap(10, &max_idx_to_segment_size);
ASSERT_EQ(3, max_idx_to_segment_size.size());
- max_idx_to_segment_size.clear();
// Check that even when the min index is the last index from the oldest segment,
// we still return 3.
log_->GetMaxIndexesToSegmentSizeMap(14, &max_idx_to_segment_size);
ASSERT_EQ(3, max_idx_to_segment_size.size());
- max_idx_to_segment_size.clear();
// Check that if the first segment is GCable, we get 2 back.
log_->GetMaxIndexesToSegmentSizeMap(15, &max_idx_to_segment_size);
ASSERT_EQ(2, max_idx_to_segment_size.size());
- max_idx_to_segment_size.clear();
// Check that if the min index is at the very end of the only segment we can get rid of that we
// get 1 back.
log_->GetMaxIndexesToSegmentSizeMap(24, &max_idx_to_segment_size);
ASSERT_EQ(1, max_idx_to_segment_size.size());
- max_idx_to_segment_size.clear();
// Check that we don't get anything back when there's nothing we want to get rid of.
log_->GetMaxIndexesToSegmentSizeMap(25, &max_idx_to_segment_size);
@@ -1043,12 +1014,6 @@ TEST_F(LogTest, TestGetMaxIndexesToSegmentSizeMap) {
// we get 0 segments back.
log_->GetMaxIndexesToSegmentSizeMap(35, &max_idx_to_segment_size);
ASSERT_EQ(0, max_idx_to_segment_size.size());
-
- // Check that logs that would normally count for log retention won't be returned since they are
- // too young.
- FLAGS_log_min_seconds_to_retain = 500;
- log_->GetMaxIndexesToSegmentSizeMap(10, &max_idx_to_segment_size);
- ASSERT_EQ(0, max_idx_to_segment_size.size());
}
// Regression test. Check that failed preallocation returns an error instead of
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/log.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log.cc b/src/kudu/consensus/log.cc
index 2e2e9f5..c5a9344 100644
--- a/src/kudu/consensus/log.cc
+++ b/src/kudu/consensus/log.cc
@@ -19,6 +19,7 @@
#include <algorithm>
#include <mutex>
+#include <limits>
#include "kudu/common/wire_protocol.h"
#include "kudu/consensus/log_index.h"
@@ -58,16 +59,13 @@ DEFINE_int32(log_min_segments_to_retain, 2,
TAG_FLAG(log_min_segments_to_retain, runtime);
TAG_FLAG(log_min_segments_to_retain, advanced);
-DEFINE_int32(log_min_seconds_to_retain, 300,
- "The minimum number of seconds for which to keep log segments to keep at all times, "
- "regardless of what is required for durability. Logs may be still retained for "
- "a longer amount of time if they are necessary for correct restart. This should be "
- "set long enough such that a tablet server which has temporarily failed can be "
- "restarted within the given time period. If a server is down for longer than this "
- "amount of time, it is possible that its tablets will be re-replicated on other "
- "machines.");
-TAG_FLAG(log_min_seconds_to_retain, runtime);
-TAG_FLAG(log_min_seconds_to_retain, advanced);
+DEFINE_int32(log_max_segments_to_retain, 10,
+ "The maximum number of past log segments to keep at all times for "
+ "the purposes of catching up other peers.");
+TAG_FLAG(log_max_segments_to_retain, runtime);
+TAG_FLAG(log_max_segments_to_retain, advanced);
+TAG_FLAG(log_max_segments_to_retain, experimental);
+
// Group commit configuration.
// -----------------------------
@@ -629,45 +627,39 @@ Status Log::Sync() {
return Status::OK();
}
-Status Log::GetSegmentsToGCUnlocked(int64_t min_op_idx, SegmentSequence* segments_to_gc) const {
- // Find the prefix of segments in the segment sequence that is guaranteed not to include
- // 'min_op_idx'.
- RETURN_NOT_OK(reader_->GetSegmentPrefixNotIncluding(min_op_idx, segments_to_gc));
-
- int max_to_delete = std::max(reader_->num_segments() - FLAGS_log_min_segments_to_retain, 0);
- if (segments_to_gc->size() > max_to_delete) {
- VLOG(2) << "GCing " << segments_to_gc->size() << " in " << log_dir_
- << " would not leave enough remaining segments to satisfy minimum "
- << "retention requirement. Only considering "
- << max_to_delete << "/" << reader_->num_segments();
- segments_to_gc->resize(max_to_delete);
- } else if (segments_to_gc->size() < max_to_delete) {
- int extra_segments = max_to_delete - segments_to_gc->size();
- VLOG(2) << tablet_id_ << " has too many log segments, need to GC "
- << extra_segments << " more. ";
- }
+int GetPrefixSizeToGC(RetentionIndexes retention, const SegmentSequence& segments) {
+ int rem_segs = segments.size();
+ int prefix_size = 0;
+ for (const scoped_refptr<ReadableLogSegment>& segment : segments) {
+ if (rem_segs <= FLAGS_log_min_segments_to_retain) {
+ break;
+ }
+
+ if (!segment->HasFooter()) break;
+
+ int64_t seg_max_idx = segment->footer().max_replicate_index();
+ // If removing this segment would compromise durability, we cannot remove it.
+ if (seg_max_idx >= retention.for_durability) {
+ break;
+ }
- // Don't GC segments that are newer than the configured time-based retention.
- int64_t now = GetCurrentTimeMicros();
- for (int i = 0; i < segments_to_gc->size(); i++) {
- const scoped_refptr<ReadableLogSegment>& segment = (*segments_to_gc)[i];
-
- // Segments here will always have a footer, since we don't return the in-progress segment
- // up above. However, segments written by older Kudu builds may not have the timestamp
- // info. In that case, we're allowed to GC them.
- if (!segment->footer().has_close_timestamp_micros()) continue;
-
- int64_t age_seconds = (now - segment->footer().close_timestamp_micros()) / 1000000;
- if (age_seconds < FLAGS_log_min_seconds_to_retain) {
- VLOG(2) << "Segment " << segment->path() << " is only " << age_seconds << "s old: "
- << "cannot GC it yet due to configured time-based retention policy.";
- // Truncate the list of segments to GC here -- if this one is too new, then
- // all later ones are also too new.
- segments_to_gc->resize(i);
+ // Check if removing this segment would compromise the ability to catch up a peer,
+ // we should retain it, unless this would break the max_segments flag.
+ if (seg_max_idx >= retention.for_peers &&
+ rem_segs <= FLAGS_log_max_segments_to_retain) {
break;
}
+
+ prefix_size++;
+ rem_segs--;
}
+ return prefix_size;
+}
+Status Log::GetSegmentsToGCUnlocked(RetentionIndexes retention,
+ SegmentSequence* segments_to_gc) const {
+ RETURN_NOT_OK(reader_->GetSegmentsSnapshot(segments_to_gc));
+ segments_to_gc->resize(GetPrefixSizeToGC(retention, *segments_to_gc));
return Status::OK();
}
@@ -709,10 +701,12 @@ void Log::GetLatestEntryOpId(consensus::OpId* op_id) const {
}
}
-Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
- CHECK_GE(min_op_idx, 0);
+Status Log::GC(RetentionIndexes retention, int32_t* num_gced) {
+ CHECK_GE(retention.for_durability, 0);
- VLOG(1) << "Running Log GC on " << log_dir_ << ": retaining ops >= " << min_op_idx;
+ VLOG(1) << "Running Log GC on " << log_dir_ << ": retaining "
+ "ops >= " << retention.for_durability << " for durability, "
+ "ops >= " << retention.for_peers << " for peers";
VLOG_TIMING(1, "Log GC") {
SegmentSequence segments_to_delete;
@@ -720,7 +714,7 @@ Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
std::lock_guard<percpu_rwlock> l(state_lock_);
CHECK_EQ(kLogWriting, log_state_);
- GetSegmentsToGCUnlocked(min_op_idx, &segments_to_delete);
+ RETURN_NOT_OK(GetSegmentsToGCUnlocked(retention, &segments_to_delete));
if (segments_to_delete.size() == 0) {
VLOG(1) << "No segments to delete.";
@@ -736,8 +730,14 @@ Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
// Now that they are no longer referenced by the Log, delete the files.
*num_gced = 0;
for (const scoped_refptr<ReadableLogSegment>& segment : segments_to_delete) {
- LOG(INFO) << "Deleting log segment in path: " << segment->path()
- << " (GCed ops < " << min_op_idx << ")";
+ string ops_str;
+ if (segment->HasFooter() && segment->footer().has_min_replicate_index()) {
+ DCHECK(segment->footer().has_max_replicate_index());
+ ops_str = Substitute(" (ops $0-$1)",
+ segment->footer().min_replicate_index(),
+ segment->footer().max_replicate_index());
+ }
+ LOG(INFO) << "Deleting log segment in path: " << segment->path() << ops_str;
RETURN_NOT_OK(fs_manager_->env()->DeleteFile(segment->path()));
(*num_gced)++;
}
@@ -752,14 +752,14 @@ Status Log::GC(int64_t min_op_idx, int32_t* num_gced) {
return Status::OK();
}
-void Log::GetGCableDataSize(int64_t min_op_idx, int64_t* total_size) const {
- CHECK_GE(min_op_idx, 0);
+void Log::GetGCableDataSize(RetentionIndexes retention, int64_t* total_size) const {
+ CHECK_GE(retention.for_durability, 0);
SegmentSequence segments_to_delete;
*total_size = 0;
{
shared_lock<rw_spinlock> l(state_lock_.get_lock());
CHECK_EQ(kLogWriting, log_state_);
- Status s = GetSegmentsToGCUnlocked(min_op_idx, &segments_to_delete);
+ Status s = GetSegmentsToGCUnlocked(retention, &segments_to_delete);
if (!s.ok() || segments_to_delete.size() == 0) {
return;
@@ -770,21 +770,24 @@ void Log::GetGCableDataSize(int64_t min_op_idx, int64_t* total_size) const {
}
}
-void Log::GetMaxIndexesToSegmentSizeMap(int64_t min_op_idx,
+void Log::GetMaxIndexesToSegmentSizeMap(int64_t idx_for_durability,
std::map<int64_t, int64_t>* max_idx_to_segment_size)
const {
- shared_lock<rw_spinlock> l(state_lock_.get_lock());
- CHECK_EQ(kLogWriting, log_state_);
- // We want to retain segments so we're only asking the extra ones.
- int segments_count = std::max(reader_->num_segments() - FLAGS_log_min_segments_to_retain, 0);
- if (segments_count == 0) {
- return;
+ max_idx_to_segment_size->clear();
+ SegmentSequence segments;
+ {
+ shared_lock<rw_spinlock> l(state_lock_.get_lock());
+ CHECK_EQ(kLogWriting, log_state_);
+ CHECK_OK(reader_->GetSegmentsSnapshot(&segments));
}
- int64_t now = GetCurrentTimeMicros();
- int64_t max_close_time_us = now - (FLAGS_log_min_seconds_to_retain * 1000000);
- reader_->GetMaxIndexesToSegmentSizeMap(min_op_idx, segments_count, max_close_time_us,
- max_idx_to_segment_size);
+ int gc_prefix = GetPrefixSizeToGC(RetentionIndexes(idx_for_durability),
+ segments);
+ for (int i = gc_prefix; i < segments.size() - FLAGS_log_min_segments_to_retain; i++) {
+ if (!segments[i]->HasFooter()) break;
+ int64_t max_repl_idx = segments[i]->footer().max_replicate_index();
+ (*max_idx_to_segment_size)[max_repl_idx] = segments[i]->file_size();
+ }
}
void Log::SetSchemaForNextLogSegment(const Schema& schema,
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/log.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log.h b/src/kudu/consensus/log.h
index e2a3e86..49623e4 100644
--- a/src/kudu/consensus/log.h
+++ b/src/kudu/consensus/log.h
@@ -18,6 +18,7 @@
#ifndef KUDU_CONSENSUS_LOG_H_
#define KUDU_CONSENSUS_LOG_H_
+#include <limits>
#include <map>
#include <memory>
#include <string>
@@ -46,6 +47,7 @@ namespace log {
struct LogEntryBatchLogicalSize;
struct LogMetrics;
+struct RetentionIndexes;
class LogEntryBatch;
class LogIndex;
class LogReader;
@@ -193,17 +195,17 @@ class Log : public RefCountedThreadSafe<Log> {
// If successful, num_gced is set to the number of deleted log segments.
//
// This method is thread-safe.
- Status GC(int64_t min_op_idx, int* num_gced);
+ Status GC(RetentionIndexes indexes, int* num_gced);
// Computes the amount of bytes that would have been GC'd if Log::GC had been called.
- void GetGCableDataSize(int64_t min_op_idx, int64_t* total_size) const;
+ void GetGCableDataSize(RetentionIndexes indexes, int64_t* total_size) const;
// Returns a map of log index -> segment size, of all the segments that currently cannot be GCed
- // because in-memory structures have anchors in them.
- //
- // 'min_op_idx' is the minimum operation index to start looking from, meaning that we skip the
- // segment that contains it and then start recording segments.
- void GetMaxIndexesToSegmentSizeMap(int64_t min_op_idx,
+ // because of an anchor on the given 'idx_for_durability' log index. Note that, even if
+ // these segments are being retained for peer catchup, they are treated as 'GCable' here,
+ // since the purpose of this method is to bound startup time by flushing in-memory stores
+ // which refer to operations far back in the log.
+ void GetMaxIndexesToSegmentSizeMap(int64_t idx_for_durability,
std::map<int64_t, int64_t>* max_idx_to_segment_size) const;
// Returns the file system location of the currently active WAL segment.
@@ -299,8 +301,8 @@ class Log : public RefCountedThreadSafe<Log> {
Status Sync();
- // Helper method to get the segment sequence to GC based on the provided min_op_idx.
- Status GetSegmentsToGCUnlocked(int64_t min_op_idx, SegmentSequence* segments_to_gc) const;
+ // Helper method to get the segment sequence to GC based on the provided 'retention' struct.
+ Status GetSegmentsToGCUnlocked(RetentionIndexes retention, SegmentSequence* segments_to_gc) const;
LogEntryBatchQueue* entry_queue() {
return &entry_batch_queue_;
@@ -398,6 +400,29 @@ class Log : public RefCountedThreadSafe<Log> {
DISALLOW_COPY_AND_ASSIGN(Log);
};
+// Indicates which log indexes should be retained for different purposes.
+//
+// When default-constructed, starts with maximum indexes, indicating no
+// logs need to be retained for either purposes.
+struct RetentionIndexes {
+ RetentionIndexes(int64_t durability = std::numeric_limits<int64_t>::max(),
+ int64_t peers = std::numeric_limits<int64_t>::max())
+ : for_durability(durability),
+ for_peers(peers) {}
+
+ // The minimum log entry index which *must* be retained in order to
+ // preserve durability and the ability to restart the local node
+ // from its WAL.
+ int64_t for_durability;
+
+ // The minimum log entry index which *should* be retained in order to
+ // catch up other peers hosting this same tablet. These entries may
+ // still be GCed in the case that they are from very old log segments
+ // or the log has become too large.
+ int64_t for_peers;
+};
+
+
// This class represents a batch of operations to be written and
// synced to the log. It is opaque to the user and is managed by the
// Log class.
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/log_reader.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log_reader.cc b/src/kudu/consensus/log_reader.cc
index 39a8c2d..6b78f8a 100644
--- a/src/kudu/consensus/log_reader.cc
+++ b/src/kudu/consensus/log_reader.cc
@@ -194,29 +194,6 @@ Status LogReader::InitEmptyReaderForTests() {
return Status::OK();
}
-Status LogReader::GetSegmentPrefixNotIncluding(int64_t index,
- SegmentSequence* segments) const {
- DCHECK_GE(index, 0);
- DCHECK(segments);
- segments->clear();
-
- std::lock_guard<simple_spinlock> lock(lock_);
- CHECK_EQ(state_, kLogReaderReading);
-
- for (const scoped_refptr<ReadableLogSegment>& segment : segments_) {
- // The last segment doesn't have a footer. Never include that one.
- if (!segment->HasFooter()) {
- break;
- }
- if (segment->footer().max_replicate_index() >= index) {
- break;
- }
- // TODO: tests for edge cases here with backwards ordered replicates.
- segments->push_back(segment);
- }
-
- return Status::OK();
-}
int64_t LogReader::GetMinReplicateIndex() const {
std::lock_guard<simple_spinlock> lock(lock_);
@@ -233,32 +210,6 @@ int64_t LogReader::GetMinReplicateIndex() const {
return min_remaining_op_idx;
}
-void LogReader::GetMaxIndexesToSegmentSizeMap(int64_t min_op_idx, int32_t segments_count,
- int64_t max_close_time_us,
- std::map<int64_t, int64_t>*
- max_idx_to_segment_size) const {
- std::lock_guard<simple_spinlock> lock(lock_);
- DCHECK_GE(segments_count, 0);
- for (const scoped_refptr<ReadableLogSegment>& segment : segments_) {
- if (max_idx_to_segment_size->size() == segments_count) {
- break;
- }
- DCHECK(segment->HasFooter());
- if (segment->footer().max_replicate_index() < min_op_idx) {
- // This means we found a log we can GC. Adjust the expected number of logs.
- segments_count--;
- continue;
- }
-
- if (max_close_time_us < segment->footer().close_timestamp_micros()) {
- int64_t age_seconds = segment->footer().close_timestamp_micros() / 1000000;
- VLOG(2) << "Segment " << segment->path() << " is only " << age_seconds << "s old: "
- << "won't be counted towards log retention";
- break;
- }
- (*max_idx_to_segment_size)[segment->footer().max_replicate_index()] = segment->file_size();
- }
-}
scoped_refptr<ReadableLogSegment> LogReader::GetSegmentBySequenceNumber(int64_t seq) const {
std::lock_guard<simple_spinlock> lock(lock_);
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/log_reader.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log_reader.h b/src/kudu/consensus/log_reader.h
index d17f041..b4666a7 100644
--- a/src/kudu/consensus/log_reader.h
+++ b/src/kudu/consensus/log_reader.h
@@ -64,31 +64,10 @@ class LogReader {
const scoped_refptr<MetricEntity>& metric_entity,
std::shared_ptr<LogReader>* reader);
- // Returns the biggest prefix of segments, from the current sequence, guaranteed
- // not to include any replicate messages with indexes >= 'index'.
- Status GetSegmentPrefixNotIncluding(int64_t index,
- SegmentSequence* segments) const;
-
// Return the minimum replicate index that is retained in the currently available
// logs. May return -1 if no replicates have been logged.
int64_t GetMinReplicateIndex() const;
- // Returns a map of maximum log index in segment -> segment size representing all the segments
- // that start after 'min_op_idx', up to 'segments_count'.
- //
- // 'min_op_idx' is the minimum operation index to start looking from, we don't record
- // the segments before the one that contain that id.
- //
- // 'segments_count' is the number of segments we'll add to the map. It _must_ be sized so that
- // we don't add the last segment. If we find logs that can be GCed, we'll decrease the number of
- // elements we'll add to the map by 1 since they.
- //
- // 'max_close_time_us' is the timestamp in microseconds from which we don't want to evict,
- // meaning that log segments that we closed after that time must not be added to the map.
- void GetMaxIndexesToSegmentSizeMap(int64_t min_op_idx, int32_t segments_count,
- int64_t max_close_time_us,
- std::map<int64_t, int64_t>* max_idx_to_segment_size) const;
-
// Return a readable segment with the given sequence number, or NULL if it
// cannot be found (e.g. if it has already been GCed).
scoped_refptr<ReadableLogSegment> GetSegmentBySequenceNumber(int64_t seq) const;
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/raft_consensus-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus-test.cc b/src/kudu/consensus/raft_consensus-test.cc
index 3ba3d64..cbbf333 100644
--- a/src/kudu/consensus/raft_consensus-test.cc
+++ b/src/kudu/consensus/raft_consensus-test.cc
@@ -323,6 +323,7 @@ ConsensusRequestPB RaftConsensusTest::MakeConsensusRequest(int64_t caller_term,
request.set_caller_term(caller_term);
request.set_caller_uuid(caller_uuid);
request.set_tablet_id(kTestTablet);
+ request.set_all_replicated_index(0);
*request.mutable_preceding_id() = preceding_opid;
return request;
}
@@ -500,6 +501,7 @@ TEST_F(RaftConsensusTest, TestAbortOperations) {
const string PEER_0_UUID = "peer-0";
request.set_caller_uuid(PEER_0_UUID);
request.set_tablet_id(kTestTablet);
+ request.set_all_replicated_index(0);
request.mutable_preceding_id()->CopyFrom(MakeOpId(2, 4));
ReplicateMsg* replicate = request.add_ops();
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/raft_consensus.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.cc b/src/kudu/consensus/raft_consensus.cc
index 6a4195a..534a594 100644
--- a/src/kudu/consensus/raft_consensus.cc
+++ b/src/kudu/consensus/raft_consensus.cc
@@ -878,7 +878,8 @@ Status RaftConsensus::CheckLeaderRequestUnlocked(const ConsensusRequestPB* reque
ConsensusResponsePB* response,
LeaderRequest* deduped_req) {
- if (request->has_deprecated_committed_index()) {
+ if (request->has_deprecated_committed_index() ||
+ !request->has_all_replicated_index()) {
return Status::InvalidArgument("Leader appears to be running an earlier version "
"of Kudu. Please shut down and upgrade all servers "
"before restarting.");
@@ -1205,7 +1206,7 @@ Status RaftConsensus::UpdateReplica(const ConsensusRequestPB* request,
VLOG_WITH_PREFIX_UNLOCKED(1) << "Marking committed up to " << apply_up_to;
TRACE("Marking committed up to $0", apply_up_to);
CHECK_OK(state_->AdvanceCommittedIndexUnlocked(apply_up_to));
- queue_->UpdateFollowerCommittedIndex(apply_up_to);
+ queue_->UpdateFollowerWatermarks(apply_up_to, request->all_replicated_index());
// We can now update the last received watermark.
//
@@ -1874,6 +1875,15 @@ Status RaftConsensus::GetLastOpId(OpIdType type, OpId* id) {
return Status::OK();
}
+log::RetentionIndexes RaftConsensus::GetRetentionIndexes() {
+ // Grab the watermarks from the queue. It's OK to fetch these two watermarks
+ // separately -- the worst case is we see a relatively "out of date" watermark
+ // which just means we'll retain slightly more than necessary in this invocation
+ // of log GC.
+ return log::RetentionIndexes(queue_->GetCommittedIndex(), // for durability
+ queue_->GetAllReplicatedIndex()); // for peers
+}
+
void RaftConsensus::MarkDirty(const std::string& reason) {
WARN_NOT_OK(thread_pool_->SubmitClosure(Bind(mark_dirty_clbk_, reason)),
state_->LogPrefixThreadSafe() + "Unable to run MarkDirty callback");
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/raft_consensus.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.h b/src/kudu/consensus/raft_consensus.h
index 451d5f1..02d6593 100644
--- a/src/kudu/consensus/raft_consensus.h
+++ b/src/kudu/consensus/raft_consensus.h
@@ -169,6 +169,8 @@ class RaftConsensus : public Consensus,
int64_t term,
const std::string& reason) override;
+ log::RetentionIndexes GetRetentionIndexes() override;
+
protected:
// Trigger that a non-Transaction ConsensusRound has finished replication.
// If the replication was successful, an status will be OK. Otherwise, it
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/consensus/raft_consensus_quorum-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus_quorum-test.cc b/src/kudu/consensus/raft_consensus_quorum-test.cc
index 2e7a746..24428fc 100644
--- a/src/kudu/consensus/raft_consensus_quorum-test.cc
+++ b/src/kudu/consensus/raft_consensus_quorum-test.cc
@@ -972,6 +972,7 @@ TEST_F(RaftConsensusQuorumTest, TestReplicasEnforceTheLogMatchingProperty) {
req.set_caller_term(last_op_id.term());
req.mutable_preceding_id()->CopyFrom(last_op_id);
req.set_committed_index(last_op_id.index());
+ req.set_all_replicated_index(0);
ReplicateMsg* replicate = req.add_ops();
replicate->set_timestamp(clock_->Now().ToUint64());
@@ -1110,6 +1111,7 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
req.set_caller_term(last_op_id.term());
req.set_caller_uuid("peer-0");
req.set_committed_index(last_op_id.index());
+ req.set_all_replicated_index(0);
ConsensusResponsePB res;
Status s = peer->Update(&req, &res);
ASSERT_EQ(last_op_id.term() + 3, res.responder_term());
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/integration-tests/delete_table-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/delete_table-test.cc b/src/kudu/integration-tests/delete_table-test.cc
index 4c8de2a..6bb3499 100644
--- a/src/kudu/integration-tests/delete_table-test.cc
+++ b/src/kudu/integration-tests/delete_table-test.cc
@@ -135,7 +135,7 @@ Status DeleteTableTest::CheckTabletTombstonedOrDeletedOnTS(
IsSuperBlockExpected is_superblock_expected) {
CHECK(data_state == TABLET_DATA_TOMBSTONED || data_state == TABLET_DATA_DELETED) << data_state;
// There should be no WALs and no cmeta.
- if (inspect_->CountWALSegmentsForTabletOnTS(index, tablet_id) > 0) {
+ if (inspect_->CountFilesInWALDirForTS(index, tablet_id) > 0) {
return Status::IllegalState("WAL segments exist for tablet", tablet_id);
}
if (is_cmeta_expected == CMETA_EXPECTED &&
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/integration-tests/external_mini_cluster_fs_inspector.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster_fs_inspector.cc b/src/kudu/integration-tests/external_mini_cluster_fs_inspector.cc
index dd6dcf4..19262ca 100644
--- a/src/kudu/integration-tests/external_mini_cluster_fs_inspector.cc
+++ b/src/kudu/integration-tests/external_mini_cluster_fs_inspector.cc
@@ -23,6 +23,7 @@
#include "kudu/consensus/metadata.pb.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/gutil/strings/util.h"
#include "kudu/fs/fs_manager.h"
#include "kudu/integration-tests/external_mini_cluster.h"
#include "kudu/util/env.h"
@@ -63,14 +64,17 @@ Status ExternalMiniClusterFsInspector::ListFilesInDir(const string& path,
return Status::OK();
}
-int ExternalMiniClusterFsInspector::CountFilesInDir(const string& path) {
+int ExternalMiniClusterFsInspector::CountFilesInDir(const string& path,
+ StringPiece pattern) {
vector<string> entries;
Status s = ListFilesInDir(path, &entries);
if (!s.ok()) return 0;
- return entries.size();
+ return std::count_if(entries.begin(), entries.end(), [&](const string& s) {
+ return pattern.empty() || MatchPattern(s, pattern);
+ });
}
-int ExternalMiniClusterFsInspector::CountWALSegmentsOnTS(int index) {
+int ExternalMiniClusterFsInspector::CountWALFilesOnTS(int index) {
string data_dir = cluster_->tablet_server(index)->data_dir();
string ts_wal_dir = JoinPathSegments(data_dir, FsManager::kWalDirName);
vector<string> tablets;
@@ -108,15 +112,17 @@ vector<string> ExternalMiniClusterFsInspector::ListTabletsWithDataOnTS(int index
return tablets;
}
-int ExternalMiniClusterFsInspector::CountWALSegmentsForTabletOnTS(int index,
- const string& tablet_id) {
+int ExternalMiniClusterFsInspector::CountFilesInWALDirForTS(
+ int index,
+ const string& tablet_id,
+ StringPiece pattern) {
string data_dir = cluster_->tablet_server(index)->data_dir();
string wal_dir = JoinPathSegments(data_dir, FsManager::kWalDirName);
string tablet_wal_dir = JoinPathSegments(wal_dir, tablet_id);
if (!env_->FileExists(tablet_wal_dir)) {
return 0;
}
- return CountFilesInDir(tablet_wal_dir);
+ return CountFilesInDir(tablet_wal_dir, pattern);
}
bool ExternalMiniClusterFsInspector::DoesConsensusMetaExistForTabletOnTS(int index,
@@ -144,7 +150,7 @@ Status ExternalMiniClusterFsInspector::CheckNoDataOnTS(int index) {
if (CountFilesInDir(JoinPathSegments(data_dir, FsManager::kTabletMetadataDirName)) > 0) {
return Status::IllegalState("tablet metadata blocks still exist", data_dir);
}
- if (CountWALSegmentsOnTS(index) > 0) {
+ if (CountWALFilesOnTS(index) > 0) {
return Status::IllegalState("wals still exist", data_dir);
}
if (CountFilesInDir(JoinPathSegments(data_dir, FsManager::kConsensusMetadataDirName)) > 0) {
@@ -257,7 +263,7 @@ Status ExternalMiniClusterFsInspector::WaitForMinFilesInTabletWalDirOnTS(int ind
int seen = 0;
MonoTime deadline = MonoTime::Now() + timeout;
while (true) {
- seen = CountWALSegmentsForTabletOnTS(index, tablet_id);
+ seen = CountFilesInWALDirForTS(index, tablet_id);
if (seen >= count) return Status::OK();
if (deadline < MonoTime::Now()) {
break;
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/integration-tests/external_mini_cluster_fs_inspector.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster_fs_inspector.h b/src/kudu/integration-tests/external_mini_cluster_fs_inspector.h
index 600e30d..6616498 100644
--- a/src/kudu/integration-tests/external_mini_cluster_fs_inspector.h
+++ b/src/kudu/integration-tests/external_mini_cluster_fs_inspector.h
@@ -22,6 +22,7 @@
#include <vector>
#include "kudu/gutil/macros.h"
+#include "kudu/gutil/strings/stringpiece.h"
#include "kudu/tablet/metadata.pb.h"
#include "kudu/util/monotime.h"
@@ -50,8 +51,9 @@ class ExternalMiniClusterFsInspector {
~ExternalMiniClusterFsInspector();
Status ListFilesInDir(const std::string& path, std::vector<std::string>* entries);
- int CountFilesInDir(const std::string& path);
- int CountWALSegmentsOnTS(int index);
+
+ // If provided, files are filtered by the glob-style pattern 'pattern'.
+ int CountFilesInDir(const std::string& path, StringPiece pattern = StringPiece());
// List all of the tablets with tablet metadata in the cluster.
std::vector<std::string> ListTablets();
@@ -64,7 +66,12 @@ class ExternalMiniClusterFsInspector {
// evidenced by their having a WAL). This excludes those that are tombstoned.
std::vector<std::string> ListTabletsWithDataOnTS(int index);
- int CountWALSegmentsForTabletOnTS(int index, const std::string& tablet_id);
+ // Return the number of files in the WAL directory for the given 'tablet_id' on TS 'index'.
+ // If provided, files are filtered by the glob-style pattern 'pattern'.
+ int CountFilesInWALDirForTS(int index,
+ const std::string& tablet_id,
+ StringPiece pattern = StringPiece());
+
bool DoesConsensusMetaExistForTabletOnTS(int index, const std::string& tablet_id);
int CountReplicasInMetadataDirs();
@@ -110,6 +117,10 @@ class ExternalMiniClusterFsInspector {
const MonoDelta& timeout = MonoDelta::FromSeconds(30));
private:
+ // Return the number of files in WAL directories on the given tablet server.
+ // This includes log index files (not just segments).
+ int CountWALFilesOnTS(int index);
+
std::string GetConsensusMetadataPathOnTS(int index,
const std::string& tablet_id) const;
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/integration-tests/raft_consensus-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/raft_consensus-itest.cc b/src/kudu/integration-tests/raft_consensus-itest.cc
index 63be00b..22a7ada 100644
--- a/src/kudu/integration-tests/raft_consensus-itest.cc
+++ b/src/kudu/integration-tests/raft_consensus-itest.cc
@@ -387,7 +387,7 @@ void RaftConsensusITest::AddFlagsForLogRolls(vector<string>* extra_tserver_flags
extra_tserver_flags->push_back("--log_segment_size_mb=1");
extra_tserver_flags->push_back("--log_async_preallocate_segments=false");
extra_tserver_flags->push_back("--log_min_segments_to_retain=1");
- extra_tserver_flags->push_back("--log_min_seconds_to_retain=0");
+ extra_tserver_flags->push_back("--log_max_segments_to_retain=3");
extra_tserver_flags->push_back("--maintenance_manager_polling_interval_ms=100");
}
@@ -615,10 +615,20 @@ void RaftConsensusITest::Write128KOpsToLeader(int num_writes) {
// Test that when a follower is stopped for a long time, the log cache
// properly evicts operations, but still allows the follower to catch
// up when it comes back.
+//
+// Also asserts that the other replicas retain logs for the stopped
+// follower to catch up from.
TEST_F(RaftConsensusITest, TestCatchupAfterOpsEvicted) {
- vector<string> extra_flags;
- extra_flags.push_back("--log_cache_size_limit_mb=1");
- extra_flags.push_back("--consensus_max_batch_size_bytes=500000");
+ vector<string> extra_flags = {
+ "--log_cache_size_limit_mb=1",
+ "--consensus_max_batch_size_bytes=500000",
+ // Use short and synchronous rolls so that we can test log segment retention.
+ "--log_segment_size_mb=1",
+ "--log_async_preallocate_segments=false",
+ // Run the maintenance manager frequently so that we don't have to wait
+ // long for GC.
+ "--maintenance_manager_polling_interval_ms=100"
+ };
BuildAndStart(extra_flags);
TServerDetails* replica = (*tablet_replicas_.begin()).second;
ASSERT_TRUE(replica != nullptr);
@@ -628,14 +638,40 @@ TEST_F(RaftConsensusITest, TestCatchupAfterOpsEvicted) {
ASSERT_OK(replica_ets->Pause());
LOG(INFO)<< "Paused one of the replicas, starting to write.";
- // Insert 3MB worth of data.
- const int kNumWrites = 25;
+ // Insert 5MB worth of data.
+ const int kNumWrites = 40;
NO_FATALS(Write128KOpsToLeader(kNumWrites));
+ // Sleep a bit to give the maintenance manager time to GC logs, if it were
+ // going to.
+ SleepFor(MonoDelta::FromSeconds(1));
+
+ // Check that the leader and non-paused follower have not GCed any logs (since
+ // the third peer needs them to catch up).
+ for (int i = 0; i < cluster_->num_tablet_servers(); i++) {
+ int num_wals = inspect_->CountFilesInWALDirForTS(i, tablet_id_, "wal-*");
+ if (cluster_->tablet_server(i) == replica_ets) {
+ ASSERT_EQ(1, num_wals) << "Replica should have only one segment";
+ } else {
+ ASSERT_EQ(6, num_wals)
+ << "Other nodes should retain segments for the frozen replica to catch up";
+ }
+ }
+
// Now unpause the replica, the lagging replica should eventually catch back up.
ASSERT_OK(replica_ets->Resume());
ASSERT_ALL_REPLICAS_AGREE(kNumWrites);
+
+ // Once the follower has caught up, all replicas should eventually GC the earlier
+ // log segments that they were retaining.
+ AssertEventually([&]() {
+ for (int i = 0; i < cluster_->num_tablet_servers(); i++) {
+ SCOPED_TRACE(Substitute("TS $0", i));
+ int num_wals = inspect_->CountFilesInWALDirForTS(i, tablet_id_, "wal-*");
+ ASSERT_EQ(2, num_wals);
+ }
+ });
}
void RaftConsensusITest::CauseFollowerToFallBehindLogGC(string* leader_uuid,
@@ -1176,6 +1212,7 @@ TEST_F(RaftConsensusITest, TestReplicaBehaviorViaRPC) {
req.set_dest_uuid(replica_ts->uuid());
req.set_caller_uuid("fake_caller");
req.set_caller_term(2);
+ req.set_all_replicated_index(0);
req.set_committed_index(1);
req.mutable_preceding_id()->CopyFrom(MakeOpId(1, 1));
@@ -2036,6 +2073,7 @@ TEST_F(RaftConsensusITest, TestEarlyCommitDespiteMemoryPressure) {
req.set_caller_uuid(tservers[2]->instance_id.permanent_uuid());
req.set_caller_term(1);
req.set_committed_index(1);
+ req.set_all_replicated_index(0);
req.mutable_preceding_id()->CopyFrom(MakeOpId(1, 1));
for (int i = 0; i < kNumOps; i++) {
AddOp(MakeOpId(1, 2 + i), &req);
@@ -2449,8 +2487,8 @@ TEST_F(RaftConsensusITest, TestEvictAbandonedFollowers) {
ASSERT_OK(WaitForServersToAgree(timeout, active_tablet_servers, tablet_id_, 2));
}
-// Test that followers that fall behind the leader's log GC threshold are
-// evicted from the config.
+// Test that, after followers are evicted from the config, the master re-adds a new
+// replica for that follower and it eventually catches back up.
TEST_F(RaftConsensusITest, TestMasterReplacesEvictedFollowers) {
vector<string> extra_flags;
AddFlagsForLogRolls(&extra_flags); // For CauseFollowerToFallBehindLogGC().
@@ -2542,6 +2580,7 @@ TEST_F(RaftConsensusITest, TestUpdateConsensusErrorNonePrepared) {
req.set_caller_uuid(tservers[2]->instance_id.permanent_uuid());
req.set_caller_term(0);
req.set_committed_index(0);
+ req.set_all_replicated_index(0);
req.mutable_preceding_id()->CopyFrom(MakeOpId(0, 0));
for (int i = 0; i < kNumOps; i++) {
AddOp(MakeOpId(0, 1 + i), &req);
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/tablet/tablet_peer-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_peer-test.cc b/src/kudu/tablet/tablet_peer-test.cc
index 4c1b004..fd6a3c8 100644
--- a/src/kudu/tablet/tablet_peer-test.cc
+++ b/src/kudu/tablet/tablet_peer-test.cc
@@ -47,7 +47,6 @@
METRIC_DECLARE_entity(tablet);
-DECLARE_int32(log_min_seconds_to_retain);
DECLARE_int32(flush_threshold_mb);
namespace kudu {
@@ -241,26 +240,15 @@ class TabletPeerTest : public KuduTabletTest {
void AssertNoLogAnchors() {
// Make sure that there are no registered anchors in the registry
CHECK_EQ(0, tablet_peer_->log_anchor_registry()->GetAnchorCountForTests());
- int64_t earliest_index = -1;
- // And that there are no in-flight transactions (which are implicit
- // anchors) by comparing the TabletPeer's earliest needed OpId and the last
- // entry in the log; if they match there is nothing in flight.
- tablet_peer_->GetEarliestNeededLogIndex(&earliest_index);
- OpId last_log_opid;
- tablet_peer_->log_->GetLatestEntryOpId(&last_log_opid);
- CHECK_EQ(earliest_index, last_log_opid.index())
- << "Found unexpected anchor: " << earliest_index
- << " Last log entry: " << last_log_opid.ShortDebugString();
}
// Assert that the Log GC() anchor is earlier than the latest OpId in the Log.
void AssertLogAnchorEarlierThanLogLatest() {
- int64_t earliest_index = -1;
- tablet_peer_->GetEarliestNeededLogIndex(&earliest_index);
+ log::RetentionIndexes retention = tablet_peer_->GetRetentionIndexes();
OpId last_log_opid;
tablet_peer_->log_->GetLatestEntryOpId(&last_log_opid);
- CHECK_LT(earliest_index, last_log_opid.index())
- << "Expected valid log anchor, got earliest opid: " << earliest_index
+ CHECK_LT(retention.for_durability, last_log_opid.index())
+ << "Expected valid log anchor, got earliest opid: " << retention.for_durability
<< " (expected any value earlier than last log id: " << last_log_opid.ShortDebugString()
<< ")";
}
@@ -304,7 +292,6 @@ class DelayedApplyTransaction : public WriteTransaction {
// Ensure that Log::GC() doesn't delete logs when the MRS has an anchor.
TEST_F(TabletPeerTest, TestMRSAnchorPreventsLogGC) {
- FLAGS_log_min_seconds_to_retain = 0;
ConsensusBootstrapInfo info;
ASSERT_OK(StartPeer(info));
@@ -325,10 +312,9 @@ TEST_F(TabletPeerTest, TestMRSAnchorPreventsLogGC) {
ASSERT_GT(tablet_peer_->log_anchor_registry()->GetAnchorCountForTests(), 0);
// Ensure nothing gets deleted.
- int64_t min_log_index = -1;
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
- ASSERT_EQ(0, num_gced) << "earliest needed: " << min_log_index;
+ log::RetentionIndexes retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
+ ASSERT_EQ(0, num_gced) << "earliest needed: " << retention.for_durability;
// Flush MRS as needed to ensure that we don't have OpId anchors in the MRS.
tablet_peer_->tablet()->Flush();
@@ -337,16 +323,15 @@ TEST_F(TabletPeerTest, TestMRSAnchorPreventsLogGC) {
// The first two segments should be deleted.
// The last is anchored due to the commit in the last segment being the last
// OpId in the log.
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
- ASSERT_EQ(2, num_gced) << "earliest needed: " << min_log_index;
+ retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
+ ASSERT_EQ(2, num_gced) << "earliest needed: " << retention.for_durability;
ASSERT_OK(log->reader()->GetSegmentsSnapshot(&segments));
ASSERT_EQ(2, segments.size());
}
// Ensure that Log::GC() doesn't delete logs when the DMS has an anchor.
TEST_F(TabletPeerTest, TestDMSAnchorPreventsLogGC) {
- FLAGS_log_min_seconds_to_retain = 0;
ConsensusBootstrapInfo info;
ASSERT_OK(StartPeer(info));
@@ -365,9 +350,8 @@ TEST_F(TabletPeerTest, TestDMSAnchorPreventsLogGC) {
// Flush MRS & GC log so the next mutation goes into a DMS.
ASSERT_OK(tablet_peer_->tablet()->Flush());
- int64_t min_log_index = -1;
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
+ log::RetentionIndexes retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
// We will only GC 1, and have 1 left because the earliest needed OpId falls
// back to the latest OpId written to the Log if no anchors are set.
ASSERT_EQ(1, num_gced);
@@ -403,8 +387,8 @@ TEST_F(TabletPeerTest, TestDMSAnchorPreventsLogGC) {
// Ensure the delta and last insert remain in the logs, anchored by the delta.
// Note that this will allow GC of the 2nd insert done above.
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
+ retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
ASSERT_EQ(1, num_gced);
ASSERT_OK(log->reader()->GetSegmentsSnapshot(&segments));
ASSERT_EQ(5, segments.size());
@@ -419,8 +403,8 @@ TEST_F(TabletPeerTest, TestDMSAnchorPreventsLogGC) {
// The last log OpId is the commit in the last segment, so it only anchors
// that segment, not the previous, because it's not the first OpId in the
// segment.
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
+ retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
ASSERT_EQ(3, num_gced);
ASSERT_OK(log->reader()->GetSegmentsSnapshot(&segments));
ASSERT_EQ(2, segments.size());
@@ -428,7 +412,6 @@ TEST_F(TabletPeerTest, TestDMSAnchorPreventsLogGC) {
// Ensure that Log::GC() doesn't compact logs with OpIds of active transactions.
TEST_F(TabletPeerTest, TestActiveTransactionPreventsLogGC) {
- FLAGS_log_min_seconds_to_retain = 0;
ConsensusBootstrapInfo info;
ASSERT_OK(StartPeer(info));
@@ -495,9 +478,8 @@ TEST_F(TabletPeerTest, TestActiveTransactionPreventsLogGC) {
// because we are delaying the Commit message with the CountDownLatch.
// GC the first four segments created by the inserts.
- int64_t min_log_index = -1;
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
+ log::RetentionIndexes retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
ASSERT_EQ(4, num_gced);
ASSERT_OK(log->reader()->GetSegmentsSnapshot(&segments));
ASSERT_EQ(2, segments.size());
@@ -515,8 +497,8 @@ TEST_F(TabletPeerTest, TestActiveTransactionPreventsLogGC) {
AssertLogAnchorEarlierThanLogLatest();
// Try to GC(), nothing should be deleted due to the in-flight transaction.
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
+ retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
ASSERT_EQ(0, num_gced);
ASSERT_OK(log->reader()->GetSegmentsSnapshot(&segments));
ASSERT_EQ(5, segments.size());
@@ -532,8 +514,8 @@ TEST_F(TabletPeerTest, TestActiveTransactionPreventsLogGC) {
AssertNoLogAnchors();
// All should be deleted except the two last segments.
- tablet_peer_->GetEarliestNeededLogIndex(&min_log_index);
- ASSERT_OK(log->GC(min_log_index, &num_gced));
+ retention = tablet_peer_->GetRetentionIndexes();
+ ASSERT_OK(log->GC(retention, &num_gced));
ASSERT_EQ(3, num_gced);
ASSERT_OK(log->reader()->GetSegmentsSnapshot(&segments));
ASSERT_EQ(2, segments.size());
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/tablet/tablet_peer.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_peer.cc b/src/kudu/tablet/tablet_peer.cc
index 24c720f..ae6ac91 100644
--- a/src/kudu/tablet/tablet_peer.cc
+++ b/src/kudu/tablet/tablet_peer.cc
@@ -364,10 +364,9 @@ Status TabletPeer::RunLogGC() {
if (!CheckRunning().ok()) {
return Status::OK();
}
- int64_t min_log_index;
int32_t num_gced;
- GetEarliestNeededLogIndex(&min_log_index);
- Status s = log_->GC(min_log_index, &num_gced);
+ log::RetentionIndexes retention = GetRetentionIndexes();
+ Status s = log_->GC(retention, &num_gced);
if (!s.ok()) {
s = s.CloneAndPrepend("Unexpected error while running Log GC from TabletPeer");
LOG(ERROR) << s.ToString();
@@ -438,18 +437,15 @@ void TabletPeer::GetInFlightTransactions(Transaction::TraceType trace_type,
}
}
-void TabletPeer::GetEarliestNeededLogIndex(int64_t* min_index) const {
- // First, we anchor on the last OpId in the Log to establish a lower bound
- // and avoid racing with the other checks. This limits the Log GC candidate
- // segments before we check the anchors.
- {
- OpId last_log_op;
- log_->GetLatestEntryOpId(&last_log_op);
- *min_index = last_log_op.index();
- }
+log::RetentionIndexes TabletPeer::GetRetentionIndexes() const {
+ // Let consensus set a minimum index that should be anchored.
+ // This ensures that we:
+ // (a) don't GC any operations which are still in flight
+ // (b) don't GC any operations that are needed to catch up lagging peers.
+ log::RetentionIndexes ret = consensus_->GetRetentionIndexes();
// If we never have written to the log, no need to proceed.
- if (*min_index == 0) return;
+ if (ret.for_durability == 0) return ret;
// Next, we interrogate the anchor registry.
// Returns OK if minimum known, NotFound if no anchors are registered.
@@ -459,7 +455,7 @@ void TabletPeer::GetEarliestNeededLogIndex(int64_t* min_index) const {
if (PREDICT_FALSE(!s.ok())) {
DCHECK(s.IsNotFound()) << "Unexpected error calling LogAnchorRegistry: " << s.ToString();
} else {
- *min_index = std::min(*min_index, min_anchor_index);
+ ret.for_durability = std::min(ret.for_durability, min_anchor_index);
}
}
@@ -471,24 +467,23 @@ void TabletPeer::GetEarliestNeededLogIndex(int64_t* min_index) const {
// A transaction which doesn't have an opid hasn't been submitted for replication yet and
// thus has no need to anchor the log.
if (tx_op_id.IsInitialized()) {
- *min_index = std::min(*min_index, tx_op_id.index());
+ ret.for_durability = std::min(ret.for_durability, tx_op_id.index());
}
}
+
+ return ret;
}
Status TabletPeer::GetMaxIndexesToSegmentSizeMap(MaxIdxToSegmentSizeMap* idx_size_map) const {
RETURN_NOT_OK(CheckRunning());
- int64_t min_op_idx;
- GetEarliestNeededLogIndex(&min_op_idx);
- log_->GetMaxIndexesToSegmentSizeMap(min_op_idx, idx_size_map);
+ log::RetentionIndexes retention = GetRetentionIndexes();
+ log_->GetMaxIndexesToSegmentSizeMap(retention.for_durability, idx_size_map);
return Status::OK();
}
Status TabletPeer::GetGCableDataSize(int64_t* retention_size) const {
RETURN_NOT_OK(CheckRunning());
- int64_t min_op_idx;
- GetEarliestNeededLogIndex(&min_op_idx);
- log_->GetGCableDataSize(min_op_idx, retention_size);
+ log_->GetGCableDataSize(GetRetentionIndexes(), retention_size);
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/e21523f5/src/kudu/tablet/tablet_peer.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_peer.h b/src/kudu/tablet/tablet_peer.h
index 8658bf0..8d72e72 100644
--- a/src/kudu/tablet/tablet_peer.h
+++ b/src/kudu/tablet/tablet_peer.h
@@ -202,9 +202,9 @@ class TabletPeer : public RefCountedThreadSafe<TabletPeer>,
void GetInFlightTransactions(Transaction::TraceType trace_type,
std::vector<consensus::TransactionStatusPB>* out) const;
- // Returns the minimum known log index that is in-memory or in-flight.
+ // Returns the log indexes to be retained for durability and to catch up peers.
// Used for selection of log segments to delete during Log GC.
- void GetEarliestNeededLogIndex(int64_t* log_index) const;
+ log::RetentionIndexes GetRetentionIndexes() const;
// Returns a map of log index -> segment size, of all the segments that currently cannot be GCed
// because in-memory structures have anchors in them.
[3/4] kudu git commit: KUDU-236 (part 1). Implement tablet history GC
Posted by to...@apache.org.
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet_history_gc-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_history_gc-test.cc b/src/kudu/tablet/tablet_history_gc-test.cc
new file mode 100644
index 0000000..02a7b3e
--- /dev/null
+++ b/src/kudu/tablet/tablet_history_gc-test.cc
@@ -0,0 +1,511 @@
+// 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 <atomic>
+#include <gflags/gflags.h>
+
+#include "kudu/server/hybrid_clock.h"
+#include "kudu/tablet/compaction.h"
+#include "kudu/tablet/mvcc.h"
+#include "kudu/tablet/tablet.h"
+#include "kudu/tablet/tablet_metrics.h"
+#include "kudu/tablet/tablet-test-base.h"
+
+DECLARE_int32(tablet_history_max_age_sec);
+DECLARE_bool(use_mock_wall_clock);
+
+using kudu::server::HybridClock;
+
+// Specify row regex to match on. Empty string means don't match anything.
+#define ASSERT_DEBUG_DUMP_ROWS_MATCH(pattern) do { \
+ const std::string& _pat = (pattern); \
+ vector<string> _rows; \
+ ASSERT_OK(tablet()->DebugDump(&_rows)); \
+ /* Ignore the non-data (formattting) lines in the output. */ \
+ std::string _base_pat = R"(^Dumping|^-|^MRS|^RowSet)"; \
+ if (!_pat.empty()) _base_pat += "|"; \
+ ASSERT_STRINGS_ALL_MATCH(_rows, _base_pat + _pat); \
+} while (0)
+
+namespace kudu {
+namespace tablet {
+
+class TabletHistoryGcTest : public TabletTestBase<IntKeyTestSetup<INT64>> {
+ public:
+ typedef TabletTestBase<IntKeyTestSetup<INT64>> Superclass;
+
+ TabletHistoryGcTest()
+ : Superclass(TabletHarness::Options::HYBRID_CLOCK) {
+ FLAGS_use_mock_wall_clock = true;
+ }
+
+ virtual void SetUp() OVERRIDE {
+ NO_FATALS(TabletTestBase<IntKeyTestSetup<INT64>>::SetUp());
+ // Mock clock defaults to 0 and this screws up the AHM calculation which ends up negative.
+ down_cast<HybridClock*>(clock())->SetMockClockWallTimeForTests(GetCurrentTimeMicros());
+ }
+
+ protected:
+ enum ToFlush {
+ FLUSH,
+ NO_FLUSH
+ };
+
+ void InsertOriginalRows(int num_rowsets, uint64_t rows_per_rowset);
+ void AddTimeToHybridClock(MonoDelta delta) {
+ uint64_t now = HybridClock::GetPhysicalValueMicros(clock()->Now());
+ uint64_t new_time = now + delta.ToMicroseconds();
+ down_cast<HybridClock*>(clock())->SetMockClockWallTimeForTests(new_time);
+ }
+
+ int64_t TotalNumRows() const { return num_rowsets_ * rows_per_rowset_; }
+
+ TestRowVerifier GenRowsEqualVerifier(int32_t expected_val) {
+ return [=](int32_t key, int32_t val) -> bool { return val == expected_val; };
+ }
+
+ const TestRowVerifier kRowsEqual0 = GenRowsEqualVerifier(0);
+ const TestRowVerifier kRowsEqual1 = GenRowsEqualVerifier(1);
+ const TestRowVerifier kRowsEqual2 = GenRowsEqualVerifier(2);
+
+ const int kStartRow = 0;
+ int num_rowsets_ = 3;
+ int rows_per_rowset_ = 300;
+};
+
+void TabletHistoryGcTest::InsertOriginalRows(int num_rowsets, uint64_t rows_per_rowset) {
+ ClampRowCount(num_rowsets * rows_per_rowset);
+ for (int rowset_id = 0; rowset_id < num_rowsets; rowset_id++) {
+ InsertTestRows(rowset_id * rows_per_rowset, rows_per_rowset, 0);
+ ASSERT_OK(tablet()->Flush());
+ }
+ ASSERT_EQ(num_rowsets, tablet()->num_rowsets());
+}
+
+// Test that we do not generate undos for redo operations that are older than
+// the AHM during major delta compaction.
+TEST_F(TabletHistoryGcTest, TestNoGenerateUndoOnMajorDeltaCompaction) {
+ FLAGS_tablet_history_max_age_sec = 1; // Keep history for 1 second.
+
+ NO_FATALS(InsertOriginalRows(num_rowsets_, rows_per_rowset_));
+ NO_FATALS(VerifyTestRowsWithVerifier(kStartRow, TotalNumRows(), kRowsEqual0));
+ Timestamp time_after_insert = clock()->Now();
+
+ // Timestamps recorded after each round of updates.
+ Timestamp post_update_ts[2];
+
+ // Mutate all of the rows, setting val=1. Then again for val=2.
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int val = 1; val <= 2; val++) {
+ for (int row_idx = 0; row_idx < TotalNumRows(); row_idx++) {
+ ASSERT_OK(UpdateTestRow(&writer, row_idx, val));
+ }
+ // We must flush the DMS before major compaction can operate on these REDOs.
+ for (int i = 0; i < num_rowsets_; i++) {
+ tablet()->FlushBiggestDMS();
+ }
+ post_update_ts[val - 1] = clock()->Now();
+ }
+
+ // Move the AHM beyond our mutations, which are represented as REDOs.
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(2)));
+
+ // Current-time reads should give us 2, but reads from the past should give
+ // us 0 or 1.
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(),
+ time_after_insert, kRowsEqual0));
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(),
+ post_update_ts[0], kRowsEqual1));
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(),
+ post_update_ts[1], kRowsEqual2));
+ NO_FATALS(VerifyTestRowsWithVerifier(kStartRow, TotalNumRows(), kRowsEqual2));
+
+ // Run major delta compaction.
+ for (int i = 0; i < num_rowsets_; i++) {
+ ASSERT_OK(tablet()->CompactWorstDeltas(RowSet::MAJOR_DELTA_COMPACTION));
+ }
+
+ // Now, we should have base data = 2 with no other historical values.
+ // Major delta compaction will not remove UNDOs, so we expect a single UNDO DELETE as well.
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=2\) Undos: \[@[[:digit:]]+\(DELETE\)\] Redos: \[\]$)");
+}
+
+// Test that major delta compaction works when run on a subset of columns:
+// 1. Insert rows and flush to DiskRowSets.
+// 2. Mutate two columns.
+// 3. Move time forward.
+// 4. Run major delta compaction on a single column.
+// 5. Make sure we don't lose anything unexpected.
+TEST_F(TabletHistoryGcTest, TestMajorDeltaCompactionOnSubsetOfColumns) {
+ FLAGS_tablet_history_max_age_sec = 100;
+
+ num_rowsets_ = 3;
+ rows_per_rowset_ = 20;
+
+ NO_FATALS(InsertOriginalRows(num_rowsets_, rows_per_rowset_));
+ NO_FATALS(VerifyTestRowsWithVerifier(kStartRow, TotalNumRows(), kRowsEqual0));
+
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int32_t row_key = 0; row_key < TotalNumRows(); row_key++) {
+ KuduPartialRow row(&client_schema_);
+ setup_.BuildRowKey(&row, row_key);
+ ASSERT_OK_FAST(row.SetInt32(1, 1));
+ ASSERT_OK_FAST(row.SetInt32(2, 2));
+ ASSERT_OK_FAST(writer.Update(row));
+ }
+ for (int i = 0; i < num_rowsets_; i++) {
+ tablet()->FlushBiggestDMS();
+ }
+
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+
+ vector<std::shared_ptr<RowSet>> rowsets;
+ tablet()->GetRowSetsForTests(&rowsets);
+ for (int i = 0; i < num_rowsets_; i++) {
+ DiskRowSet* drs = down_cast<DiskRowSet*>(rowsets[i].get());
+ vector<ColumnId> col_ids_to_compact = { schema_.column_id(2) };
+ ASSERT_OK(drs->MajorCompactDeltaStoresWithColumnIds(col_ids_to_compact,
+ tablet()->GetHistoryGcOpts()));
+ }
+
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=2\) Undos: \[@[[:digit:]]+\(DELETE\)\] )"
+ R"(Redos: \[@[[:digit:]]+\(SET key_idx=1\)\]$)");
+
+ vector<string> rows;
+ ASSERT_OK(IterateToStringList(&rows));
+ ASSERT_EQ(TotalNumRows(), rows.size());
+}
+
+// Tests the following two MRS flush scenarios:
+// 1. Verify that no UNDO is generated after inserting a row into the MRS,
+// waiting for the AHM to pass, then flushing the MRS.
+// 2. Same as #1 but delete the inserted row from the MRS before waiting for
+// the AHM to pass.
+TEST_F(TabletHistoryGcTest, TestNoGenerateUndoOnMRSFlush) {
+ FLAGS_tablet_history_max_age_sec = 100;
+
+ Timestamp time_before_insert = clock()->Now();
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int32_t i = kStartRow; i < TotalNumRows(); i++) {
+ ASSERT_OK(InsertTestRow(&writer, i, 0));
+ }
+ Timestamp time_after_insert = clock()->Now();
+ for (int32_t i = kStartRow; i < TotalNumRows(); i++) {
+ ASSERT_OK(DeleteTestRow(&writer, i));
+ }
+ Timestamp time_after_delete = clock()->Now();
+
+ // Move the clock.
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, 0,
+ time_before_insert, boost::none));
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(),
+ time_after_insert, kRowsEqual0));
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, 0,
+ time_after_delete, boost::none));
+
+ // Now flush the MRS. No trace should remain after this.
+ ASSERT_OK(tablet()->Flush());
+ ASSERT_DEBUG_DUMP_ROWS_MATCH("");
+
+ for (const auto& rsmd : tablet()->metadata()->rowsets()) {
+ ASSERT_EQ(0, rsmd->undo_delta_blocks().size());
+ }
+ ASSERT_EQ(0, tablet()->EstimateOnDiskSize());
+
+ // Now check the same thing (flush not generating an UNDO), but without the
+ // delete following the insert. We do it with a single row.
+
+ ASSERT_OK(InsertTestRow(&writer, kStartRow, 0));
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+ ASSERT_OK(tablet()->Flush());
+ // There should be no undo blocks, despite flushing an insert.
+ for (const auto& rsmd : tablet()->metadata()->rowsets()) {
+ ASSERT_EQ(0, rsmd->undo_delta_blocks().size());
+ }
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, 1, Timestamp(0), kRowsEqual0));
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=0\) Undos: \[\] Redos: \[\]$)");
+}
+
+// Test that undos get GCed on a merge compaction.
+// In this test, we GC the UNDO that undoes the insert.
+TEST_F(TabletHistoryGcTest, TestUndoGCOnMergeCompaction) {
+ FLAGS_tablet_history_max_age_sec = 1; // Keep history for 1 second.
+
+ Timestamp time_before_insert = clock()->Now();
+ NO_FATALS(InsertOriginalRows(num_rowsets_, rows_per_rowset_));
+ NO_FATALS(VerifyTestRowsWithVerifier(kStartRow, TotalNumRows(), kRowsEqual0));
+
+ // The earliest thing we can see is an empty tablet.
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, 0, time_before_insert, boost::none));
+
+ // Move the clock so the insert is prior to the AHM, then compact.
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(2)));
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+
+ // Now the only thing we can see is the base data.
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(), time_before_insert,
+ kRowsEqual0));
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=0\) Undos: \[\] Redos: \[\]$)");
+}
+
+// Test that we GC the history and existence of entire deleted rows on a merge compaction.
+TEST_F(TabletHistoryGcTest, TestRowRemovalGCOnMergeCompaction) {
+ FLAGS_tablet_history_max_age_sec = 100; // Keep history for 100 seconds.
+
+ NO_FATALS(InsertOriginalRows(num_rowsets_, rows_per_rowset_));
+ NO_FATALS(VerifyTestRowsWithVerifier(kStartRow, TotalNumRows(), kRowsEqual0));
+
+ Timestamp prev_time = clock()->Now();
+
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+
+ // Delete all of the rows in the tablet.
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int row_idx = 0; row_idx < TotalNumRows(); row_idx++) {
+ ASSERT_OK(DeleteTestRow(&writer, row_idx));
+ }
+ ASSERT_OK(tablet()->Flush());
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(
+ R"(int32 val=0\) Undos: \[@[[:digit:]]+\(DELETE\)\] Redos: \[@[[:digit:]]+\(DELETE\)\]$)");
+
+ // Compaction at this time will only remove the initial UNDO records. The
+ // DELETE REDOs are too recent.
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=0\) Undos: \[\] Redos: \[@[[:digit:]]+\(DELETE\)\]$)");
+
+ // Move the AHM so that the delete is now prior to it.
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+
+ // Now that even the deletion is prior to the AHM, all of the on-disk data
+ // will be GCed.
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+ ASSERT_DEBUG_DUMP_ROWS_MATCH("");
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, 0, prev_time, boost::none));
+ ASSERT_EQ(0, tablet()->EstimateOnDiskSize());
+}
+
+// Test that we don't over-aggressively GC history prior to the AHM.
+TEST_F(TabletHistoryGcTest, TestNoUndoGCUntilAncientHistoryMark) {
+ FLAGS_tablet_history_max_age_sec = 1000; // 1000 seconds before we GC history.
+
+ NO_FATALS(InsertOriginalRows(num_rowsets_, rows_per_rowset_));
+
+ Timestamp prev_time = clock()->Now();
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(2)));
+
+ // Mutate all of the rows.
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int row_idx = 0; row_idx < TotalNumRows(); row_idx++) {
+ SCOPED_TRACE(Substitute("Row index: $0", row_idx));
+ ASSERT_OK(UpdateTestRow(&writer, row_idx, 1));
+ }
+
+ // Current-time reads should give us 1, but reads from the past should give us 0.
+ NO_FATALS(VerifyTestRowsWithVerifier(kStartRow, TotalNumRows(), kRowsEqual1));
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(), prev_time,
+ kRowsEqual0));
+
+ for (int i = 0; i < num_rowsets_; i++) {
+ ASSERT_OK(tablet()->CompactWorstDeltas(RowSet::MINOR_DELTA_COMPACTION));
+ ASSERT_OK(tablet()->CompactWorstDeltas(RowSet::MAJOR_DELTA_COMPACTION));
+ }
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+
+ // Still read 0 from the past.
+ NO_FATALS(VerifyTestRowsWithTimestampAndVerifier(kStartRow, TotalNumRows(), prev_time,
+ kRowsEqual0));
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=1\) Undos: \[@[[:digit:]]+\(SET val=0\), )"
+ R"(@[[:digit:]]+\(DELETE\)\] Redos: \[\]$)");
+}
+
+// Test that "ghost" rows (deleted on one rowset, reinserted on another) don't
+// get revived after history GC.
+TEST_F(TabletHistoryGcTest, TestGhostRowsNotRevived) {
+ FLAGS_tablet_history_max_age_sec = 100;
+
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int i = 0; i <= 2; i++) {
+ ASSERT_OK(InsertTestRow(&writer, 0, i));
+ ASSERT_OK(DeleteTestRow(&writer, 0));
+ ASSERT_OK(tablet()->Flush());
+ }
+
+ // Create one more rowset on disk which has just an INSERT (ie a non-ghost row).
+ ASSERT_OK(InsertTestRow(&writer, 0, 3));
+ ASSERT_OK(tablet()->Flush());
+
+ // Move the clock, then compact. This should result in a rowset with just one
+ // row in it.
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+
+ // We should end up with a single row as base data.
+ NO_FATALS(VerifyTestRows(0, 1));
+ ASSERT_DEBUG_DUMP_ROWS_MATCH(R"(int32 val=3\) Undos: \[\] Redos: \[\])");
+}
+
+// Test to ensure that nothing bad happens when we partially GC different rows
+// in a rowset. We delete alternating keys to end up with a mix of GCed and
+// non-GCed rows in each rowset.
+TEST_F(TabletHistoryGcTest, TestGcOnAlternatingRows) {
+ FLAGS_tablet_history_max_age_sec = 100;
+ num_rowsets_ = 3;
+ rows_per_rowset_ = 5;
+
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int rowset_id = 0; rowset_id < num_rowsets_; rowset_id++) {
+ for (int i = 0; i < rows_per_rowset_; i++) {
+ int32_t row_key = rowset_id * rows_per_rowset_ + i;
+ ASSERT_OK(InsertTestRow(&writer, row_key, 0));
+ }
+ ASSERT_OK(tablet()->Flush());
+ }
+
+ // Delete all the odd rows.
+ for (int32_t row_key = 1; row_key < TotalNumRows(); row_key += 2) {
+ ASSERT_OK(DeleteTestRow(&writer, row_key));
+ }
+
+ // Move the clock and compact. We should end up with even rows.
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+
+ vector<string> rows;
+ ASSERT_OK(IterateToStringList(&rows));
+ ASSERT_EQ(TotalNumRows() / 2 + 1, rows.size());
+
+ // Even row keys are assigned negative values in this test framework and so
+ // end up sorted negatively.
+ std::reverse(rows.begin(), rows.end());
+
+ int i = 0;
+ for (int32_t row_key = 0; row_key < TotalNumRows(); row_key += 2) {
+ ASSERT_STR_CONTAINS(rows[i], Substitute("int64 key=$0, int32 key_idx=$1, int32 val=0",
+ -1 * row_key, row_key));
+ i++;
+ }
+}
+
+// Ensure that ReupdateMissedDeltas() doesn't reupdate the wrong row.
+// 1. Insert rows and flush.
+// 2. Delete some rows.
+// 3. Move time forward.
+// 4. Begin merge compaction.
+// 5. Insert some of the deleted rows after phase 1 snapshot is written but before phase 2.
+// 6. Update some of the rows in-between the deleted rows.
+// 7. Ensure that the rows all look right according to what we expect.
+//
+// This test uses the following pattern. Rows with even keys are deleted, rows
+// with odd keys are used in the test. The following takes place:
+// - Rows 1 and 5 are inserted with values equaling their keys and are not mutated.
+// - Rows 3 and 9 are inserted and then updated with values equaling their keys * 10 + 1.
+// - Row 7 is deleted and then reinserted, as well as updated using successive values.
+TEST_F(TabletHistoryGcTest, TestGcWithConcurrentCompaction) {
+ FLAGS_tablet_history_max_age_sec = 100;
+
+ class MyCommonHooks : public Tablet::FlushCompactCommonHooks {
+ public:
+ explicit MyCommonHooks(TabletHistoryGcTest* test)
+ : test_(test),
+ offset_(0) {
+ }
+
+ Status PostWriteSnapshot() OVERRIDE {
+ LocalTabletWriter writer(test_->tablet().get(), &test_->client_schema());
+ int offset = offset_.load(std::memory_order_acquire);
+ // Update our reinserted row.
+ CHECK_OK(test_->UpdateTestRow(&writer, 7, 73 + offset));
+
+ // Also insert and update other rows after the flush.
+ CHECK_OK(test_->UpdateTestRow(&writer, 3, 30 + offset));
+ CHECK_OK(test_->UpdateTestRow(&writer, 9, 90 + offset));
+ return Status::OK();
+ }
+
+ void set_offset(int offset) {
+ offset_.store(offset, std::memory_order_release);
+ }
+
+ private:
+ TabletHistoryGcTest* const test_;
+ std::atomic<int> offset_;
+ };
+
+ std::shared_ptr<MyCommonHooks> hooks = std::make_shared<MyCommonHooks>(this);
+ tablet()->SetFlushCompactCommonHooksForTests(hooks);
+
+ LocalTabletWriter writer(tablet().get(), &client_schema_);
+ for (int i = 0; i < 10; i++) {
+ ASSERT_OK(InsertTestRow(&writer, i, i));
+ }
+ // Also generate a reinsert.
+ ASSERT_OK(DeleteTestRow(&writer, 7));
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+ ASSERT_OK(InsertTestRow(&writer, 7, 71));
+ CHECK_OK(UpdateTestRow(&writer, 7, 72));
+
+ // Flush the rowset.
+ ASSERT_OK(tablet()->Flush());
+
+ // Delete every even row.
+ for (int i = 0; i < 10; i += 2) {
+ ASSERT_OK(DeleteTestRow(&writer, i));
+ }
+
+ NO_FATALS(AddTimeToHybridClock(MonoDelta::FromSeconds(200)));
+
+ hooks->set_offset(1);
+ ASSERT_OK(tablet()->Compact(Tablet::FORCE_COMPACT_ALL));
+ tablet()->SetFlushCompactCommonHooksForTests(nullptr);
+
+ vector<string> rows;
+ ASSERT_OK(IterateToStringList(&rows));
+
+ if (VLOG_IS_ON(2)) {
+ for (const string& r : rows) {
+ VLOG(2) << r;
+ }
+ }
+
+ vector<int32_t> expected_rows = { 1, 3, 5, 7, 9 };
+ for (int i = 0; i < expected_rows.size(); i++) {
+ int32_t key = expected_rows[i];
+ switch (key) {
+ case 1:
+ case 5:
+ ASSERT_STR_CONTAINS(rows[i], Substitute("int32 key_idx=$0, int32 val=$1",
+ key, key));
+ break;
+ case 3:
+ case 9:
+ ASSERT_STR_CONTAINS(rows[i], Substitute("int32 key_idx=$0, int32 val=$1",
+ key, key * 10 + 1));
+ break;
+ case 7:
+ ASSERT_STR_CONTAINS(rows[i], Substitute("int32 key_idx=$0, int32 val=$1",
+ key, key * 10 + 4));
+ break;
+ default:
+ break;
+ }
+ }
+}
+
+} // namespace tablet
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tserver/tablet_service.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/tablet_service.cc b/src/kudu/tserver/tablet_service.cc
index 34013a8..2a5c7cd 100644
--- a/src/kudu/tserver/tablet_service.cc
+++ b/src/kudu/tserver/tablet_service.cc
@@ -36,6 +36,7 @@
#include "kudu/rpc/rpc_context.h"
#include "kudu/rpc/rpc_sidecar.h"
#include "kudu/server/hybrid_clock.h"
+#include "kudu/tablet/compaction.h"
#include "kudu/tablet/metadata.pb.h"
#include "kudu/tablet/tablet_bootstrap.h"
#include "kudu/tablet/tablet_metrics.h"
@@ -82,6 +83,41 @@ DEFINE_int32(scanner_inject_latency_on_each_batch_ms, 0,
TAG_FLAG(scanner_inject_latency_on_each_batch_ms, unsafe);
DECLARE_int32(memory_limit_warn_threshold_percentage);
+DECLARE_int32(tablet_history_max_age_sec);
+
+using google::protobuf::RepeatedPtrField;
+using kudu::consensus::ChangeConfigRequestPB;
+using kudu::consensus::ChangeConfigResponsePB;
+using kudu::consensus::CONSENSUS_CONFIG_ACTIVE;
+using kudu::consensus::CONSENSUS_CONFIG_COMMITTED;
+using kudu::consensus::Consensus;
+using kudu::consensus::ConsensusConfigType;
+using kudu::consensus::ConsensusRequestPB;
+using kudu::consensus::ConsensusResponsePB;
+using kudu::consensus::GetLastOpIdRequestPB;
+using kudu::consensus::GetNodeInstanceRequestPB;
+using kudu::consensus::GetNodeInstanceResponsePB;
+using kudu::consensus::LeaderStepDownRequestPB;
+using kudu::consensus::LeaderStepDownResponsePB;
+using kudu::consensus::RunLeaderElectionRequestPB;
+using kudu::consensus::RunLeaderElectionResponsePB;
+using kudu::consensus::StartTabletCopyRequestPB;
+using kudu::consensus::StartTabletCopyResponsePB;
+using kudu::consensus::VoteRequestPB;
+using kudu::consensus::VoteResponsePB;
+using kudu::rpc::ResultTracker;
+using kudu::rpc::RpcContext;
+using kudu::server::HybridClock;
+using kudu::tablet::AlterSchemaTransactionState;
+using kudu::tablet::Tablet;
+using kudu::tablet::TabletPeer;
+using kudu::tablet::TabletStatusPB;
+using kudu::tablet::TransactionCompletionCallback;
+using kudu::tablet::WriteTransactionState;
+using std::shared_ptr;
+using std::unique_ptr;
+using std::vector;
+using strings::Substitute;
namespace kudu {
namespace cfile {
@@ -93,40 +129,6 @@ extern const char* CFILE_CACHE_HIT_BYTES_METRIC_NAME;
namespace kudu {
namespace tserver {
-using consensus::ChangeConfigRequestPB;
-using consensus::ChangeConfigResponsePB;
-using consensus::CONSENSUS_CONFIG_ACTIVE;
-using consensus::CONSENSUS_CONFIG_COMMITTED;
-using consensus::Consensus;
-using consensus::ConsensusConfigType;
-using consensus::ConsensusRequestPB;
-using consensus::ConsensusResponsePB;
-using consensus::GetLastOpIdRequestPB;
-using consensus::GetNodeInstanceRequestPB;
-using consensus::GetNodeInstanceResponsePB;
-using consensus::LeaderStepDownRequestPB;
-using consensus::LeaderStepDownResponsePB;
-using consensus::RunLeaderElectionRequestPB;
-using consensus::RunLeaderElectionResponsePB;
-using consensus::StartTabletCopyRequestPB;
-using consensus::StartTabletCopyResponsePB;
-using consensus::VoteRequestPB;
-using consensus::VoteResponsePB;
-
-using google::protobuf::RepeatedPtrField;
-using rpc::ResultTracker;
-using rpc::RpcContext;
-using std::shared_ptr;
-using std::unique_ptr;
-using std::vector;
-using strings::Substitute;
-using tablet::AlterSchemaTransactionState;
-using tablet::Tablet;
-using tablet::TabletPeer;
-using tablet::TabletStatusPB;
-using tablet::TransactionCompletionCallback;
-using tablet::WriteTransactionState;
-
namespace {
// Lookup the given tablet, ensuring that it both exists and is RUNNING.
@@ -1041,7 +1043,7 @@ void TabletServiceImpl::Scan(const ScanRequestPB* req,
ScanResultCopier collector(&data, rows_data.get(), indirect_data.get());
bool has_more_results = false;
- TabletServerErrorPB::Code error_code;
+ TabletServerErrorPB::Code error_code = TabletServerErrorPB::UNKNOWN_ERROR;
if (req->has_new_scan_request()) {
const NewScanRequestPB& scan_pb = req->new_scan_request();
scoped_refptr<TabletPeer> tablet_peer;
@@ -1509,6 +1511,37 @@ Status TabletServiceImpl::HandleNewScanRequest(TabletPeer* tablet_peer,
return s;
}
+ // If this is a snapshot scan and the user specified a specific timestamp to
+ // scan at, then check that we are not attempting to scan at a time earlier
+ // than the ancient history mark. Only perform this check if tablet history
+ // GC is enabled.
+ //
+ // TODO: This validation essentially prohibits scans with READ_AT_SNAPSHOT
+ // when history_max_age is set to zero. There is a tablet history GC related
+ // race when the history max age is set to very low, or zero. Imagine a case
+ // where a scan was started and READ_AT_SNAPSHOT was specified without
+ // specifying a snapshot timestamp, and --tablet_history_max_age_sec=0. The
+ // above code path will select the latest timestamp (under a lock) prior to
+ // calling RowIterator::Init(), which actually opens the blocks. That means
+ // that there is an opportunity in between those two calls for tablet history
+ // GC to kick in and delete some history. In fact, we may easily not actually
+ // end up with a valid snapshot in that case. It would be more correct to
+ // initialize the row iterator and then select the latest timestamp
+ // represented by those open files in that case.
+ Timestamp ancient_history_mark;
+ tablet::HistoryGcOpts history_gc_opts = tablet->GetHistoryGcOpts();
+ if (scan_pb.read_mode() == READ_AT_SNAPSHOT &&
+ history_gc_opts.IsAncientHistory(*snap_timestamp)) {
+ // Now that we have initialized our row iterator at a snapshot, return an
+ // error if the snapshot timestamp was prior to the ancient history mark.
+ // We have to check after we open the iterator in order to avoid a TOCTOU
+ // error.
+ *error_code = TabletServerErrorPB::INVALID_SNAPSHOT;
+ return Status::InvalidArgument("Snapshot timestamp is earlier than the ancient history mark",
+ "consider increasing the value of the configuration parameter "
+ "--tablet_history_max_age_sec");
+ }
+
*has_more_results = iter->HasNext();
TRACE("has_more: $0", *has_more_results);
if (!*has_more_results) {
[2/4] kudu git commit: Create base class for MiniCluster and
ExternalMiniCluster
Posted by to...@apache.org.
Create base class for MiniCluster and ExternalMiniCluster
This allows utility classes to interoperate with both of these types of
clusters.
Some minor changes to support this:
* Consistent interface for CreateClient()
* Consistent interface for Shutdown()
Additionally, I removed the 'readability/inheritance' rule from
'make lint' because it prevents using both 'virtual' and 'final' on the
same method, however that syntax is required by Clang in C++11 mode.
Change-Id: I62256168a9245c845e99f7253f07e69a225954bf
Reviewed-on: http://gerrit.cloudera.org:8080/3974
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/d5cc5727
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/d5cc5727
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/d5cc5727
Branch: refs/heads/master
Commit: d5cc572750cc604e9d1c0e6cd04c8f440cc8b790
Parents: e21523f
Author: Mike Percy <mp...@apache.org>
Authored: Sat Aug 13 22:07:15 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Fri Sep 9 19:13:12 2016 +0000
----------------------------------------------------------------------
build-support/lint.sh | 2 +-
src/kudu/client/client-test.cc | 4 --
src/kudu/client/predicate-test.cc | 10 +--
src/kudu/client/scan_token-test.cc | 13 +---
src/kudu/integration-tests/all_types-itest.cc | 3 +-
.../alter_table-randomized-test.cc | 3 +-
.../integration-tests/client-stress-test.cc | 6 +-
src/kudu/integration-tests/cluster_verifier.cc | 4 +-
.../integration-tests/create-table-itest.cc | 2 +-
.../external_mini_cluster-itest-base.h | 3 +-
.../integration-tests/external_mini_cluster.cc | 29 +++++---
.../integration-tests/external_mini_cluster.h | 25 +++----
.../flex_partitioning-itest.cc | 8 +--
.../full_stack-insert-scan-test.cc | 7 --
src/kudu/integration-tests/linked_list-test.cc | 5 +-
.../integration-tests/master-stress-test.cc | 2 +-
.../integration-tests/master_failover-itest.cc | 2 +-
.../integration-tests/master_migration-itest.cc | 6 +-
.../master_replication-itest.cc | 8 ---
src/kudu/integration-tests/mini_cluster.cc | 32 ++++-----
src/kudu/integration-tests/mini_cluster.h | 30 ++++-----
src/kudu/integration-tests/mini_cluster_base.h | 70 ++++++++++++++++++++
src/kudu/integration-tests/tablet_copy-itest.cc | 3 +-
src/kudu/integration-tests/test_workload.cc | 4 +-
src/kudu/integration-tests/test_workload.h | 6 +-
src/kudu/integration-tests/ts_itest-base.h | 3 -
src/kudu/integration-tests/ts_recovery-itest.cc | 3 +-
.../ts_tablet_manager-itest.cc | 6 --
src/kudu/tserver/ts_tablet_manager-test.cc | 4 ++
29 files changed, 156 insertions(+), 147 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/build-support/lint.sh
----------------------------------------------------------------------
diff --git a/build-support/lint.sh b/build-support/lint.sh
index fef47a6..d99f347 100755
--- a/build-support/lint.sh
+++ b/build-support/lint.sh
@@ -51,7 +51,7 @@ cd $ROOT
$ROOT/thirdparty/installed/bin/cpplint.py \
--verbose=4 \
- --filter=-whitespace/comments,-readability/todo,-build/header_guard,-build/include_order,-legal/copyright,-build/c++11 \
+ --filter=-whitespace/comments,-readability/todo,-readability/inheritance,-build/header_guard,-build/include_order,-legal/copyright,-build/c++11 \
$FILES 2>&1 | grep -v 'Done processing' | tee $TMP
NUM_ERRORS=$(grep "Total errors found" $TMP | awk '{print $4}')
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/client/client-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/client-test.cc b/src/kudu/client/client-test.cc
index ac50834..37a9bac 100644
--- a/src/kudu/client/client-test.cc
+++ b/src/kudu/client/client-test.cc
@@ -148,10 +148,6 @@ class ClientTest : public KuduTest {
}
void TearDown() override {
- if (cluster_) {
- cluster_->Shutdown();
- cluster_.reset();
- }
KuduTest::TearDown();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/client/predicate-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/predicate-test.cc b/src/kudu/client/predicate-test.cc
index 1a56da8..72013be 100644
--- a/src/kudu/client/predicate-test.cc
+++ b/src/kudu/client/predicate-test.cc
@@ -50,15 +50,7 @@ class PredicateTest : public KuduTest {
// Set up the mini cluster
cluster_.reset(new MiniCluster(env_.get(), MiniClusterOptions()));
ASSERT_OK(cluster_->Start());
- KuduClientBuilder client_builder;
- ASSERT_OK(cluster_->CreateClient(&client_builder, &client_));
- }
-
- void TearDown() override {
- if (cluster_) {
- cluster_->Shutdown();
- cluster_.reset();
- }
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
// Creates a key/value table schema with an int64 key and value of the
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/client/scan_token-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/scan_token-test.cc b/src/kudu/client/scan_token-test.cc
index 95de9f1..02a943a 100644
--- a/src/kudu/client/scan_token-test.cc
+++ b/src/kudu/client/scan_token-test.cc
@@ -49,15 +49,7 @@ class ScanTokenTest : public KuduTest {
// Set up the mini cluster
cluster_.reset(new MiniCluster(env_.get(), MiniClusterOptions()));
ASSERT_OK(cluster_->Start());
- KuduClientBuilder client_builder;
- ASSERT_OK(cluster_->CreateClient(&client_builder, &client_));
- }
-
- void TearDown() override {
- if (cluster_) {
- cluster_->Shutdown();
- cluster_.reset();
- }
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
// Creates a new session in manual flush mode.
@@ -79,9 +71,8 @@ class ScanTokenTest : public KuduTest {
CHECK_OK(token->Serialize(&buf));
threads.emplace_back(thread([this, &rows] (string serialized_token) {
- KuduClientBuilder client_builder;
shared_ptr<KuduClient> client;
- ASSERT_OK(cluster_->CreateClient(&client_builder, &client));
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client));
KuduScanner* scanner_ptr;
ASSERT_OK(KuduScanToken::DeserializeIntoScanner(client.get(),
serialized_token,
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/all_types-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/all_types-itest.cc b/src/kudu/integration-tests/all_types-itest.cc
index ab353c3..76d9fa1 100644
--- a/src/kudu/integration-tests/all_types-itest.cc
+++ b/src/kudu/integration-tests/all_types-itest.cc
@@ -227,8 +227,7 @@ class AllTypesItest : public KuduTest {
cluster_.reset(new ExternalMiniCluster(opts));
RETURN_NOT_OK(cluster_->Start());
- KuduClientBuilder builder;
- return cluster_->CreateClient(builder, &client_);
+ return cluster_->CreateClient(nullptr, &client_);
}
Status CreateTable() {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/alter_table-randomized-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/alter_table-randomized-test.cc b/src/kudu/integration-tests/alter_table-randomized-test.cc
index 717f0e3..55b5318 100644
--- a/src/kudu/integration-tests/alter_table-randomized-test.cc
+++ b/src/kudu/integration-tests/alter_table-randomized-test.cc
@@ -79,8 +79,7 @@ class AlterTableRandomized : public KuduTest {
cluster_.reset(new ExternalMiniCluster(opts));
ASSERT_OK(cluster_->Start());
- KuduClientBuilder builder;
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
void TearDown() override {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/client-stress-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/client-stress-test.cc b/src/kudu/integration-tests/client-stress-test.cc
index 71b8a1a..3c1a1f4 100644
--- a/src/kudu/integration-tests/client-stress-test.cc
+++ b/src/kudu/integration-tests/client-stress-test.cc
@@ -127,9 +127,8 @@ TEST_F(ClientStressTest, TestStartScans) {
// is empty.
for (int run = 1; run <= (AllowSlowTests() ? 10 : 2); run++) {
LOG(INFO) << "Starting run " << run;
- KuduClientBuilder builder;
client::sp::shared_ptr<KuduClient> client;
- CHECK_OK(cluster_->CreateClient(builder, &client));
+ CHECK_OK(cluster_->CreateClient(nullptr, &client));
CountDownLatch go_latch(1);
vector<scoped_refptr<Thread> > threads;
@@ -285,9 +284,8 @@ TEST_F(ClientStressTest_LowMemory, TestMemoryThrottling) {
TEST_F(ClientStressTest, TestUniqueClientIds) {
set<string> client_ids;
for (int i = 0; i < 1000; i++) {
- KuduClientBuilder builder;
client::sp::shared_ptr<KuduClient> client;
- CHECK_OK(cluster_->CreateClient(builder, &client));
+ CHECK_OK(cluster_->CreateClient(nullptr, &client));
string client_id = client->data_->client_id_;
auto result = client_ids.insert(client_id);
EXPECT_TRUE(result.second) << "Unique id generation failed. New client id: " << client_id
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/cluster_verifier.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/cluster_verifier.cc b/src/kudu/integration-tests/cluster_verifier.cc
index c701779..4617b54 100644
--- a/src/kudu/integration-tests/cluster_verifier.cc
+++ b/src/kudu/integration-tests/cluster_verifier.cc
@@ -107,9 +107,7 @@ Status ClusterVerifier::DoCheckRowCount(const std::string& table_name,
ComparisonMode mode,
int expected_row_count) {
client::sp::shared_ptr<client::KuduClient> client;
- client::KuduClientBuilder builder;
- RETURN_NOT_OK_PREPEND(cluster_->CreateClient(builder,
- &client),
+ RETURN_NOT_OK_PREPEND(cluster_->CreateClient(nullptr, &client),
"Unable to connect to cluster");
client::sp::shared_ptr<client::KuduTable> table;
RETURN_NOT_OK_PREPEND(client->OpenTable(table_name, &table),
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/create-table-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/create-table-itest.cc b/src/kudu/integration-tests/create-table-itest.cc
index 5a15420..18ca093 100644
--- a/src/kudu/integration-tests/create-table-itest.cc
+++ b/src/kudu/integration-tests/create-table-itest.cc
@@ -257,7 +257,7 @@ TEST_F(CreateTableITest, TestCreateTableWithDeadTServers) {
// master considers the tservers unresponsive (and recreates the
// outstanding table's tablets) during the test.
"--tserver_unresponsive_timeout_ms=5000" }));
- cluster_->Shutdown(ExternalMiniCluster::TS_ONLY);
+ cluster_->ShutdownNodes(ClusterNodes::TS_ONLY);
Schema schema(GetSimpleTestSchema());
client::KuduSchema client_schema(client::KuduSchemaFromSchema(schema));
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/external_mini_cluster-itest-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster-itest-base.h b/src/kudu/integration-tests/external_mini_cluster-itest-base.h
index e2a9cf5..d1c2167 100644
--- a/src/kudu/integration-tests/external_mini_cluster-itest-base.h
+++ b/src/kudu/integration-tests/external_mini_cluster-itest-base.h
@@ -84,8 +84,7 @@ void ExternalMiniClusterITestBase::StartCluster(const std::vector<std::string>&
ASSERT_OK(itest::CreateTabletServerMap(cluster_->master_proxy().get(),
cluster_->messenger(),
&ts_map_));
- client::KuduClientBuilder builder;
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/external_mini_cluster.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster.cc b/src/kudu/integration-tests/external_mini_cluster.cc
index d77e1d3..7a46db9 100644
--- a/src/kudu/integration-tests/external_mini_cluster.cc
+++ b/src/kudu/integration-tests/external_mini_cluster.cc
@@ -154,18 +154,20 @@ Status ExternalMiniCluster::Start() {
return Status::OK();
}
-void ExternalMiniCluster::Shutdown(NodeSelectionMode mode) {
- if (mode == ALL) {
+
+void ExternalMiniCluster::ShutdownNodes(ClusterNodes nodes) {
+ if (nodes == ClusterNodes::ALL || nodes == ClusterNodes::TS_ONLY) {
+ for (const scoped_refptr<ExternalTabletServer>& ts : tablet_servers_) {
+ ts->Shutdown();
+ }
+ }
+ if (nodes == ClusterNodes::ALL || nodes == ClusterNodes::MASTERS_ONLY) {
for (const scoped_refptr<ExternalMaster>& master : masters_) {
if (master) {
master->Shutdown();
}
}
}
-
- for (const scoped_refptr<ExternalTabletServer>& ts : tablet_servers_) {
- ts->Shutdown();
- }
}
Status ExternalMiniCluster::Restart() {
@@ -483,14 +485,19 @@ std::shared_ptr<MasterServiceProxy> ExternalMiniCluster::master_proxy(int idx) {
new MasterServiceProxy(messenger_, CHECK_NOTNULL(master(idx))->bound_rpc_addr()));
}
-Status ExternalMiniCluster::CreateClient(client::KuduClientBuilder& builder,
- client::sp::shared_ptr<client::KuduClient>* client) {
+Status ExternalMiniCluster::CreateClient(client::KuduClientBuilder* builder,
+ client::sp::shared_ptr<client::KuduClient>* client) const {
+ client::KuduClientBuilder defaults;
+ if (builder == nullptr) {
+ builder = &defaults;
+ }
+
CHECK(!masters_.empty());
- builder.clear_master_server_addrs();
+ builder->clear_master_server_addrs();
for (const scoped_refptr<ExternalMaster>& master : masters_) {
- builder.add_master_server_addr(master->bound_rpc_hostport().ToString());
+ builder->add_master_server_addr(master->bound_rpc_hostport().ToString());
}
- return builder.Build(client);
+ return builder->Build(client);
}
Status ExternalMiniCluster::SetFlag(ExternalDaemon* daemon,
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/external_mini_cluster.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/external_mini_cluster.h b/src/kudu/integration-tests/external_mini_cluster.h
index abe3f39..ed93c00 100644
--- a/src/kudu/integration-tests/external_mini_cluster.h
+++ b/src/kudu/integration-tests/external_mini_cluster.h
@@ -26,6 +26,7 @@
#include "kudu/gutil/gscoped_ptr.h"
#include "kudu/gutil/macros.h"
#include "kudu/gutil/ref_counted.h"
+#include "kudu/integration-tests/mini_cluster_base.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/status.h"
@@ -118,19 +119,13 @@ struct ExternalMiniClusterOptions {
// cluster participants, which isn't feasible in the normal MiniCluster.
// On the other hand, there is little access to inspect the internal state
// of the daemons.
-class ExternalMiniCluster {
+class ExternalMiniCluster : public MiniClusterBase {
public:
- // Mode to which node types a certain action (like Shutdown()) should apply.
- enum NodeSelectionMode {
- TS_ONLY,
- ALL
- };
-
explicit ExternalMiniCluster(const ExternalMiniClusterOptions& opts);
- ~ExternalMiniCluster();
+ virtual ~ExternalMiniCluster();
// Start the cluster.
- Status Start();
+ Status Start() override;
// Restarts the cluster. Requires that it has been Shutdown() first.
Status Restart();
@@ -144,10 +139,8 @@ class ExternalMiniCluster {
// Requires that the master is already running.
Status AddTabletServer();
- // Shuts down the whole cluster or part of it, depending on the selected
- // 'mode'.
// Currently, this uses SIGKILL on each daemon for a non-graceful shutdown.
- void Shutdown(NodeSelectionMode mode = ALL);
+ void ShutdownNodes(ClusterNodes nodes) override;
// Return the IP address that the tablet server with the given index will bind to.
// If options.bind_to_unique_loopback_addresses is false, this will be 127.0.0.1
@@ -202,11 +195,11 @@ class ExternalMiniCluster {
// Return all tablet servers and masters.
std::vector<ExternalDaemon*> daemons() const;
- int num_tablet_servers() const {
+ int num_tablet_servers() const override {
return tablet_servers_.size();
}
- int num_masters() const {
+ int num_masters() const override {
return masters_.size();
}
@@ -245,8 +238,8 @@ class ExternalMiniCluster {
// be overridden to talk to the running master.
//
// REQUIRES: the cluster must have already been Start()ed.
- Status CreateClient(client::KuduClientBuilder& builder,
- client::sp::shared_ptr<client::KuduClient>* client);
+ Status CreateClient(client::KuduClientBuilder* builder,
+ client::sp::shared_ptr<client::KuduClient>* client) const override;
// Sets the given flag on the given daemon, which must be running.
//
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/flex_partitioning-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/flex_partitioning-itest.cc b/src/kudu/integration-tests/flex_partitioning-itest.cc
index 2fe9b50..43393f7 100644
--- a/src/kudu/integration-tests/flex_partitioning-itest.cc
+++ b/src/kudu/integration-tests/flex_partitioning-itest.cc
@@ -160,13 +160,7 @@ class FlexPartitioningITest : public KuduTest {
cluster_.reset(new ExternalMiniCluster(opts));
ASSERT_OK(cluster_->Start());
- KuduClientBuilder builder;
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
- }
-
- void TearDown() override {
- cluster_->Shutdown();
- KuduTest::TearDown();
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
protected:
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/full_stack-insert-scan-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/full_stack-insert-scan-test.cc b/src/kudu/integration-tests/full_stack-insert-scan-test.cc
index 5a1a61b..2ea11c8 100644
--- a/src/kudu/integration-tests/full_stack-insert-scan-test.cc
+++ b/src/kudu/integration-tests/full_stack-insert-scan-test.cc
@@ -140,13 +140,6 @@ class FullStackInsertScanTest : public KuduTest {
ASSERT_OK(client_->OpenTable(kTableName, &reader_table_));
}
- virtual void TearDown() OVERRIDE {
- if (cluster_) {
- cluster_->Shutdown();
- }
- KuduTest::TearDown();
- }
-
void DoConcurrentClientInserts();
void DoTestScans();
void FlushToDisk();
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/linked_list-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/linked_list-test.cc b/src/kudu/integration-tests/linked_list-test.cc
index ed5574a..111d7d3 100644
--- a/src/kudu/integration-tests/linked_list-test.cc
+++ b/src/kudu/integration-tests/linked_list-test.cc
@@ -113,8 +113,7 @@ class LinkedListTest : public tserver::TabletServerIntegrationTestBase {
}
void ResetClientAndTester() {
- KuduClientBuilder builder;
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
tester_.reset(new LinkedListTester(client_, kTableId,
FLAGS_num_chains,
FLAGS_num_tablets,
@@ -124,7 +123,7 @@ class LinkedListTest : public tserver::TabletServerIntegrationTestBase {
void RestartCluster() {
CHECK(cluster_);
- cluster_->Shutdown(ExternalMiniCluster::TS_ONLY);
+ cluster_->ShutdownNodes(ClusterNodes::TS_ONLY);
cluster_->Restart();
ResetClientAndTester();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/master-stress-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/master-stress-test.cc b/src/kudu/integration-tests/master-stress-test.cc
index 0a6a579..875d901 100644
--- a/src/kudu/integration-tests/master-stress-test.cc
+++ b/src/kudu/integration-tests/master-stress-test.cc
@@ -132,7 +132,7 @@ class MasterStressTest : public KuduTest {
// Encourage the client to switch masters quickly in the event of failover.
builder.default_rpc_timeout(MonoDelta::FromSeconds(1));
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
+ ASSERT_OK(cluster_->CreateClient(&builder, &client_));
}
void TearDown() override {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/master_failover-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/master_failover-itest.cc b/src/kudu/integration-tests/master_failover-itest.cc
index 1744852..a2205c6 100644
--- a/src/kudu/integration-tests/master_failover-itest.cc
+++ b/src/kudu/integration-tests/master_failover-itest.cc
@@ -107,7 +107,7 @@ class MasterFailoverTest : public KuduTest {
// builders, but there's no such option for DeleteTable, so we extend
// the global operation timeout.
builder.default_admin_operation_timeout(MonoDelta::FromSeconds(90));
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
+ ASSERT_OK(cluster_->CreateClient(&builder, &client_));
}
Status CreateTable(const std::string& table_name, CreateTableMode mode) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/master_migration-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/master_migration-itest.cc b/src/kudu/integration-tests/master_migration-itest.cc
index 17a73ec..ee9fb59 100644
--- a/src/kudu/integration-tests/master_migration-itest.cc
+++ b/src/kudu/integration-tests/master_migration-itest.cc
@@ -79,8 +79,7 @@ class MasterMigrationTest : public KuduTest {
static Status CreateTable(ExternalMiniCluster* cluster,
const std::string& table_name) {
shared_ptr<KuduClient> client;
- KuduClientBuilder builder;
- RETURN_NOT_OK(cluster->CreateClient(builder, &client));
+ RETURN_NOT_OK(cluster->CreateClient(nullptr, &client));
KuduSchema schema;
KuduSchemaBuilder b;
b.AddColumn("key")->Type(KuduColumnSchema::INT32)->NotNull()->PrimaryKey();
@@ -190,8 +189,7 @@ TEST_F(MasterMigrationTest, TestEndToEndMigration) {
// Perform an operation that requires an elected leader.
shared_ptr<KuduClient> client;
- KuduClientBuilder builder;
- ASSERT_OK(migrated_cluster.CreateClient(builder, &client));
+ ASSERT_OK(migrated_cluster.CreateClient(nullptr, &client));
shared_ptr<KuduTable> table;
ASSERT_OK(client->OpenTable(kTableName, &table));
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/master_replication-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/master_replication-itest.cc b/src/kudu/integration-tests/master_replication-itest.cc
index 920fed0..7647420 100644
--- a/src/kudu/integration-tests/master_replication-itest.cc
+++ b/src/kudu/integration-tests/master_replication-itest.cc
@@ -70,14 +70,6 @@ class MasterReplicationTest : public KuduTest {
ASSERT_OK(cluster_->Start());
}
- virtual void TearDown() OVERRIDE {
- if (cluster_) {
- cluster_->Shutdown();
- cluster_.reset();
- }
- KuduTest::TearDown();
- }
-
Status RestartCluster() {
cluster_->Shutdown();
RETURN_NOT_OK(cluster_->Start());
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/mini_cluster.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/mini_cluster.cc b/src/kudu/integration-tests/mini_cluster.cc
index 4cf05e6..d633151 100644
--- a/src/kudu/integration-tests/mini_cluster.cc
+++ b/src/kudu/integration-tests/mini_cluster.cc
@@ -65,7 +65,7 @@ MiniCluster::MiniCluster(Env* env, const MiniClusterOptions& options)
}
MiniCluster::~MiniCluster() {
- CHECK(!running_);
+ Shutdown();
}
Status MiniCluster::Start() {
@@ -177,21 +177,20 @@ Status MiniCluster::AddTabletServer() {
return Status::OK();
}
-void MiniCluster::Shutdown() {
- for (const shared_ptr<MiniTabletServer>& tablet_server : mini_tablet_servers_) {
- tablet_server->Shutdown();
+void MiniCluster::ShutdownNodes(ClusterNodes nodes) {
+ if (nodes == ClusterNodes::ALL || nodes == ClusterNodes::TS_ONLY) {
+ for (const shared_ptr<MiniTabletServer>& tablet_server : mini_tablet_servers_) {
+ tablet_server->Shutdown();
+ }
+ mini_tablet_servers_.clear();
}
- mini_tablet_servers_.clear();
- ShutdownMasters();
- running_ = false;
-}
-
-void MiniCluster::ShutdownMasters() {
- for (shared_ptr<MiniMaster>& master_server : mini_masters_) {
- master_server->Shutdown();
- master_server.reset();
+ if (nodes == ClusterNodes::ALL || nodes == ClusterNodes::MASTERS_ONLY) {
+ for (const shared_ptr<MiniMaster>& master_server : mini_masters_) {
+ master_server->Shutdown();
+ }
+ mini_masters_.clear();
}
- mini_masters_.clear();
+ running_ = false;
}
MiniMaster* MiniCluster::mini_master(int idx) const {
@@ -279,10 +278,11 @@ Status MiniCluster::WaitForTabletServerCount(int count,
Status MiniCluster::CreateClient(KuduClientBuilder* builder,
client::sp::shared_ptr<KuduClient>* client) const {
- KuduClientBuilder default_builder;
+ client::KuduClientBuilder defaults;
if (builder == nullptr) {
- builder = &default_builder;
+ builder = &defaults;
}
+
builder->clear_master_server_addrs();
for (const shared_ptr<MiniMaster>& master : mini_masters_) {
CHECK(master);
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/mini_cluster.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/mini_cluster.h b/src/kudu/integration-tests/mini_cluster.h
index b54b17a..a04a89c 100644
--- a/src/kudu/integration-tests/mini_cluster.h
+++ b/src/kudu/integration-tests/mini_cluster.h
@@ -24,6 +24,7 @@
#include "kudu/client/shared_ptr.h"
#include "kudu/gutil/macros.h"
+#include "kudu/integration-tests/mini_cluster_base.h"
#include "kudu/util/env.h"
namespace kudu {
@@ -70,24 +71,21 @@ struct MiniClusterOptions {
// An in-process cluster with a MiniMaster and a configurable
// number of MiniTabletServers for use in tests.
-class MiniCluster {
+class MiniCluster : public MiniClusterBase {
public:
MiniCluster(Env* env, const MiniClusterOptions& options);
- ~MiniCluster();
+ virtual ~MiniCluster();
// Start a cluster with a Master and 'num_tablet_servers' TabletServers.
// All servers run on the loopback interface with ephemeral ports.
- Status Start();
+ Status Start() override;
// Like the previous method but performs initialization synchronously, i.e.
// this will wait for all TS's to be started and initialized. Tests should
// use this if they interact with tablets immediately after Start();
Status StartSync();
- void Shutdown();
-
- // Shuts down masters only.
- void ShutdownMasters();
+ void ShutdownNodes(ClusterNodes nodes) override;
// Setup a consensus configuration of distributed masters, with count specified in
// 'options'. Requires that a reserve RPC port is specified in
@@ -113,13 +111,17 @@ class MiniCluster {
master::MiniMaster* mini_master(int idx) const;
// Return number of mini masters.
- int num_masters() const { return mini_masters_.size(); }
+ int num_masters() const override {
+ return mini_masters_.size();
+ }
// Returns the TabletServer at index 'idx' of this MiniCluster.
// 'idx' must be between 0 and 'num_tablet_servers' -1.
tserver::MiniTabletServer* mini_tablet_server(int idx) const;
- int num_tablet_servers() const { return mini_tablet_servers_.size(); }
+ int num_tablet_servers() const override {
+ return mini_tablet_servers_.size();
+ }
std::string GetMasterFsRoot(int indx) const;
@@ -144,14 +146,8 @@ class MiniCluster {
Status WaitForTabletServerCount(int count, MatchMode mode,
std::vector<std::shared_ptr<master::TSDescriptor>>* descs) const;
- // Create a client configured to talk to this cluster. Builder may contain
- // override options for the client. The master address will be overridden to
- // talk to the running master. If 'builder' is NULL, default options will be
- // used.
- //
- // REQUIRES: the cluster must have already been Start()ed.
Status CreateClient(client::KuduClientBuilder* builder,
- client::sp::shared_ptr<client::KuduClient>* client) const;
+ client::sp::shared_ptr<client::KuduClient>* client) const override;
// Determine the leader master of the cluster. Sets 'idx' to the leader
// master's index (for calls to to mini_master()).
@@ -177,6 +173,8 @@ class MiniCluster {
std::vector<std::shared_ptr<master::MiniMaster> > mini_masters_;
std::vector<std::shared_ptr<tserver::MiniTabletServer> > mini_tablet_servers_;
+
+ DISALLOW_COPY_AND_ASSIGN(MiniCluster);
};
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/mini_cluster_base.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/mini_cluster_base.h b/src/kudu/integration-tests/mini_cluster_base.h
new file mode 100644
index 0000000..1fd1985
--- /dev/null
+++ b/src/kudu/integration-tests/mini_cluster_base.h
@@ -0,0 +1,70 @@
+// 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.
+#pragma once
+
+#include "kudu/client/shared_ptr.h"
+
+namespace kudu {
+class Status;
+
+namespace client {
+class KuduClient;
+class KuduClientBuilder;
+}
+
+// Mode to which node types a certain action (like Shutdown()) should apply.
+enum class ClusterNodes {
+ ALL,
+ MASTERS_ONLY,
+ TS_ONLY
+};
+
+// Base class for MiniCluster implementations. Provides some commonly-used
+// virtual methods that allow for abstracting the details of whether the
+// mini-cluster implementation is in-process or out-of-process.
+class MiniClusterBase {
+ public:
+ MiniClusterBase() {}
+ virtual ~MiniClusterBase() {}
+
+ // Start the cluster.
+ virtual Status Start() = 0;
+
+ // Convenience method for shutting down the whole cluster.
+ virtual void Shutdown() final {
+ ShutdownNodes(ClusterNodes::ALL);
+ }
+
+ // Shuts down the whole cluster or part of it, depending on the selected
+ // 'nodes'.
+ virtual void ShutdownNodes(ClusterNodes nodes) = 0;
+
+ virtual int num_masters() const = 0;
+
+ virtual int num_tablet_servers() const = 0;
+
+ // Create a client configured to talk to this cluster. 'builder' may contain
+ // override options for the client. The master address will be overridden to
+ // talk to the running master(s). If 'builder' is a nullptr, default options
+ // will be used.
+ //
+ // REQUIRES: the cluster must have already been Start()ed.
+ virtual Status CreateClient(client::KuduClientBuilder* builder,
+ client::sp::shared_ptr<client::KuduClient>* client) const = 0;
+};
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/tablet_copy-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/tablet_copy-itest.cc b/src/kudu/integration-tests/tablet_copy-itest.cc
index 43932f7..3d85aa6 100644
--- a/src/kudu/integration-tests/tablet_copy-itest.cc
+++ b/src/kudu/integration-tests/tablet_copy-itest.cc
@@ -120,8 +120,7 @@ void TabletCopyITest::StartCluster(const vector<string>& extra_tserver_flags,
ASSERT_OK(itest::CreateTabletServerMap(cluster_->master_proxy().get(),
cluster_->messenger(),
&ts_map_));
- KuduClientBuilder builder;
- ASSERT_OK(cluster_->CreateClient(builder, &client_));
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
// If a rogue (a.k.a. zombie) leader tries to replace a tombstoned
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/test_workload.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/test_workload.cc b/src/kudu/integration-tests/test_workload.cc
index c189b3c..cca1e82 100644
--- a/src/kudu/integration-tests/test_workload.cc
+++ b/src/kudu/integration-tests/test_workload.cc
@@ -50,7 +50,7 @@ using client::sp::shared_ptr;
const char* const TestWorkload::kDefaultTableName = "test-workload";
-TestWorkload::TestWorkload(ExternalMiniCluster* cluster)
+TestWorkload::TestWorkload(MiniClusterBase* cluster)
: cluster_(cluster),
payload_bytes_(11),
num_write_threads_(4),
@@ -167,7 +167,7 @@ void TestWorkload::WriteThread() {
}
void TestWorkload::Setup() {
- CHECK_OK(cluster_->CreateClient(client_builder_, &client_));
+ CHECK_OK(cluster_->CreateClient(&client_builder_, &client_));
bool table_exists;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/test_workload.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/test_workload.h b/src/kudu/integration-tests/test_workload.h
index fd3ec7e..5065858 100644
--- a/src/kudu/integration-tests/test_workload.h
+++ b/src/kudu/integration-tests/test_workload.h
@@ -29,7 +29,7 @@
namespace kudu {
-class ExternalMiniCluster;
+class MiniClusterBase;
class Thread;
// Utility class for generating a workload against a test cluster.
@@ -41,7 +41,7 @@ class TestWorkload {
public:
static const char* const kDefaultTableName;
- explicit TestWorkload(ExternalMiniCluster* cluster);
+ explicit TestWorkload(MiniClusterBase* cluster);
~TestWorkload();
void set_payload_bytes(int n) {
@@ -155,7 +155,7 @@ class TestWorkload {
private:
void WriteThread();
- ExternalMiniCluster* cluster_;
+ MiniClusterBase* cluster_;
client::KuduClientBuilder client_builder_;
client::sp::shared_ptr<client::KuduClient> client_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/ts_itest-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/ts_itest-base.h b/src/kudu/integration-tests/ts_itest-base.h
index 15e232c..4b8f905 100644
--- a/src/kudu/integration-tests/ts_itest-base.h
+++ b/src/kudu/integration-tests/ts_itest-base.h
@@ -421,9 +421,6 @@ class TabletServerIntegrationTestBase : public TabletServerTestBase {
}
virtual void TearDown() OVERRIDE {
- if (cluster_) {
- cluster_->Shutdown();
- }
STLDeleteValues(&tablet_servers_);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/ts_recovery-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/ts_recovery-itest.cc b/src/kudu/integration-tests/ts_recovery-itest.cc
index 18eee52..a3cdebf 100644
--- a/src/kudu/integration-tests/ts_recovery-itest.cc
+++ b/src/kudu/integration-tests/ts_recovery-itest.cc
@@ -312,9 +312,8 @@ TEST_P(Kudu969Test, Test) {
work.Setup();
// Open the client and table.
- KuduClientBuilder builder;
shared_ptr<KuduClient> client;
- ASSERT_OK(cluster_->CreateClient(builder, &client));
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client));
shared_ptr<KuduTable> table;
CHECK_OK(client->OpenTable(work.table_name(), &table));
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/integration-tests/ts_tablet_manager-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/ts_tablet_manager-itest.cc b/src/kudu/integration-tests/ts_tablet_manager-itest.cc
index ad78aaa..16d516d 100644
--- a/src/kudu/integration-tests/ts_tablet_manager-itest.cc
+++ b/src/kudu/integration-tests/ts_tablet_manager-itest.cc
@@ -77,7 +77,6 @@ class TsTabletManagerITest : public KuduTest {
: schema_(SimpleIntKeyKuduSchema()) {
}
virtual void SetUp() OVERRIDE;
- virtual void TearDown() OVERRIDE;
protected:
const KuduSchema schema_;
@@ -100,11 +99,6 @@ void TsTabletManagerITest::SetUp() {
ASSERT_OK(cluster_->CreateClient(nullptr, &client_));
}
-void TsTabletManagerITest::TearDown() {
- cluster_->Shutdown();
- KuduTest::TearDown();
-}
-
// Test that when the leader changes, the tablet manager gets notified and
// includes that information in the next tablet report.
TEST_F(TsTabletManagerITest, TestReportNewLeaderOnLeaderChange) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/d5cc5727/src/kudu/tserver/ts_tablet_manager-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tserver/ts_tablet_manager-test.cc b/src/kudu/tserver/ts_tablet_manager-test.cc
index 2d01868..5d55378 100644
--- a/src/kudu/tserver/ts_tablet_manager-test.cc
+++ b/src/kudu/tserver/ts_tablet_manager-test.cc
@@ -71,6 +71,10 @@ class TsTabletManagerTest : public KuduTest {
heartbeater_ = mini_server_->server()->heartbeater();
}
+ virtual void TearDown() OVERRIDE {
+ KuduTest::TearDown();
+ }
+
Status CreateNewTablet(const std::string& tablet_id,
const Schema& schema,
scoped_refptr<tablet::TabletPeer>* out_tablet_peer) {
[4/4] kudu git commit: KUDU-236 (part 1). Implement tablet history GC
Posted by to...@apache.org.
KUDU-236 (part 1). Implement tablet history GC
This patch defines the concept of an "ancient history mark" (AHM) and a
background job that removes history from disk that represents changes
occurring prior to the ancient history mark.
There is also a mechanism to reject requests for scans if they attempt
to open a snapshot scan at a timestamp prior to the AHM.
Included tests:
* Test for Redo GC via major delta compaction
* Test for history GC via tablet flush
* Test for Undo GC via merge compaction
* Test for entire-row GC via merge compaction
* Test for ghost rows in multiple RS reinsert case
* Test for reupdating missed deltas
* Test for partial rowset history GC using alternating rows
* Test for major delta compaction against a subset of columns
* Test for reinsert after delete that is prior to AHM
* Test for opening a scanner at TS < AHM and rejecting that scanner at
the RPC level
Tests coming in a follow-up commit:
* Randomized test
Missing features coming in follow-up commits:
* Background task that schedules and performs undo GC
* GC history on delta flush
Change-Id: If9833a863f118eb82be80ea56204d0d9141611c2
Reviewed-on: http://gerrit.cloudera.org:8080/3076
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/be719edc
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/be719edc
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/be719edc
Branch: refs/heads/master
Commit: be719edc3581802e094c3af6a88d67acba44ba71
Parents: d5cc572
Author: Mike Percy <mp...@apache.org>
Authored: Mon Jul 18 13:40:14 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Fri Sep 9 19:13:18 2016 +0000
----------------------------------------------------------------------
src/kudu/common/timestamp.h | 10 +
src/kudu/integration-tests/CMakeLists.txt | 1 +
.../tablet_history_gc-itest.cc | 52 ++
src/kudu/server/clock.h | 6 +
src/kudu/server/hybrid_clock.cc | 8 +-
src/kudu/server/hybrid_clock.h | 4 +-
src/kudu/tablet/CMakeLists.txt | 1 +
src/kudu/tablet/compaction-test.cc | 12 +-
src/kudu/tablet/compaction.cc | 161 ++++--
src/kudu/tablet/compaction.h | 66 ++-
src/kudu/tablet/delta_compaction.cc | 39 +-
src/kudu/tablet/delta_compaction.h | 6 +-
src/kudu/tablet/diskrowset.cc | 15 +-
src/kudu/tablet/diskrowset.h | 8 +-
src/kudu/tablet/mutation.h | 6 +-
src/kudu/tablet/rowset_metadata.h | 2 +-
src/kudu/tablet/tablet-harness.h | 16 +-
src/kudu/tablet/tablet-test-base.h | 79 ++-
src/kudu/tablet/tablet-test-util.h | 9 +-
src/kudu/tablet/tablet-test.cc | 11 +-
src/kudu/tablet/tablet.cc | 78 ++-
src/kudu/tablet/tablet.h | 11 +
src/kudu/tablet/tablet_history_gc-test.cc | 511 +++++++++++++++++++
src/kudu/tserver/tablet_service.cc | 103 ++--
24 files changed, 1047 insertions(+), 168 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/common/timestamp.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/timestamp.h b/src/kudu/common/timestamp.h
index 9c9731a..c1d46c9 100644
--- a/src/kudu/common/timestamp.h
+++ b/src/kudu/common/timestamp.h
@@ -50,8 +50,14 @@ class Timestamp {
// Encode the timestamp to the given buffer.
void EncodeTo(faststring *dst) const;
+ // Returns -1 if this < other.
+ // Returns 1 if this > other.
+ // Returns 0 if this == other.
int CompareTo(const Timestamp &other) const;
+ // Returns true if this comes before other.
+ bool ComesBefore(const Timestamp &other) const;
+
std::string ToString() const;
// Returns this Timestamp as an uint64_t
@@ -92,6 +98,10 @@ inline int Timestamp::CompareTo(const Timestamp &other) const {
return 0;
}
+inline bool Timestamp::ComesBefore(const Timestamp &other) const {
+ return v < other.v;
+}
+
} // namespace kudu
#endif /* KUDU_COMMON_TIMESTAMP_H_ */
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/integration-tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index 89b5fb8..aa6a673 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -64,6 +64,7 @@ ADD_KUDU_TEST(raft_consensus-itest RUN_SERIAL true)
ADD_KUDU_TEST(registration-test RESOURCE_LOCK "master-web-port")
ADD_KUDU_TEST(table_locations-itest)
ADD_KUDU_TEST(tablet_copy-itest)
+ADD_KUDU_TEST(tablet_history_gc-itest)
ADD_KUDU_TEST(tablet_replacement-itest)
ADD_KUDU_TEST(ts_recovery-itest)
ADD_KUDU_TEST(ts_tablet_manager-itest)
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/integration-tests/tablet_history_gc-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/tablet_history_gc-itest.cc b/src/kudu/integration-tests/tablet_history_gc-itest.cc
new file mode 100644
index 0000000..0ee404c
--- /dev/null
+++ b/src/kudu/integration-tests/tablet_history_gc-itest.cc
@@ -0,0 +1,52 @@
+// 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/integration-tests/external_mini_cluster-itest-base.h"
+#include "kudu/integration-tests/test_workload.h"
+
+using kudu::client::KuduScanner;
+using kudu::client::KuduTable;
+using kudu::client::sp::shared_ptr;
+
+namespace kudu {
+
+class TabletHistoryGcITest : public ExternalMiniClusterITestBase {
+};
+
+// Check that attempts to scan prior to the ancient history mark fail.
+TEST_F(TabletHistoryGcITest, TestSnapshotScanBeforeAHM) {
+ NO_FATALS(StartCluster({ "--tablet_history_max_age_sec=0" }));
+
+ // Create a tablet so we can scan it.
+ TestWorkload workload(cluster_.get());
+ workload.Setup();
+
+ // When the tablet history max age is set to 0, it's not possible to do a
+ // snapshot scan without a timestamp because it's illegal to open a snapshot
+ // prior to the AHM. When a snapshot timestamp is not specified, we decide on
+ // the timestamp of the snapshot before checking that it's lower than the
+ // current AHM. This test verifies that scans prior to the AHM are rejected.
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(client_->OpenTable(TestWorkload::kDefaultTableName, &table));
+ KuduScanner scanner(table.get());
+ ASSERT_OK(scanner.SetReadMode(KuduScanner::READ_AT_SNAPSHOT));
+ Status s = scanner.Open();
+ ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ ASSERT_STR_CONTAINS(s.ToString(), "Snapshot timestamp is earlier than the ancient history mark");
+}
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/server/clock.h
----------------------------------------------------------------------
diff --git a/src/kudu/server/clock.h b/src/kudu/server/clock.h
index e55fa5b..777667b 100644
--- a/src/kudu/server/clock.h
+++ b/src/kudu/server/clock.h
@@ -65,6 +65,12 @@ class Clock : public RefCountedThreadSafe<Clock> {
// Indicates whether this clock supports the required external consistency mode.
virtual bool SupportsExternalConsistencyMode(ExternalConsistencyMode mode) = 0;
+ // Indicates whether the clock has a physical component to its timestamps
+ // (wallclock time).
+ virtual bool HasPhysicalComponent() {
+ return false;
+ }
+
// Update the clock with a transaction timestamp originating from
// another server. For instance replicas can call this so that,
// if elected leader, they are guaranteed to generate timestamps
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/server/hybrid_clock.cc
----------------------------------------------------------------------
diff --git a/src/kudu/server/hybrid_clock.cc b/src/kudu/server/hybrid_clock.cc
index 6d79d1f..cc30015 100644
--- a/src/kudu/server/hybrid_clock.cc
+++ b/src/kudu/server/hybrid_clock.cc
@@ -299,6 +299,10 @@ bool HybridClock::SupportsExternalConsistencyMode(ExternalConsistencyMode mode)
return true;
}
+bool HybridClock::HasPhysicalComponent() {
+ return true;
+}
+
Status HybridClock::WaitUntilAfter(const Timestamp& then_latest,
const MonoTime& deadline) {
TRACE_EVENT0("clock", "HybridClock::WaitUntilAfter");
@@ -343,8 +347,8 @@ Status HybridClock::WaitUntilAfter(const Timestamp& then_latest,
return Status::OK();
}
- Status HybridClock::WaitUntilAfterLocally(const Timestamp& then,
- const MonoTime& deadline) {
+Status HybridClock::WaitUntilAfterLocally(const Timestamp& then,
+ const MonoTime& deadline) {
while (true) {
Timestamp now;
uint64_t error;
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/server/hybrid_clock.h
----------------------------------------------------------------------
diff --git a/src/kudu/server/hybrid_clock.h b/src/kudu/server/hybrid_clock.h
index 1110f9b..d712c82 100644
--- a/src/kudu/server/hybrid_clock.h
+++ b/src/kudu/server/hybrid_clock.h
@@ -60,6 +60,8 @@ class HybridClock : public Clock {
// HybridClock supports all external consistency modes.
virtual bool SupportsExternalConsistencyMode(ExternalConsistencyMode mode) OVERRIDE;
+ virtual bool HasPhysicalComponent() OVERRIDE;
+
// Blocks the caller thread until the true time is after 'then'.
// In other words, waits until the HybridClock::Now() on _all_ nodes
// will return a value greater than 'then'.
@@ -136,7 +138,7 @@ class HybridClock : public Clock {
uint64_t logical_value);
// Creates a new timestamp whose physical time is GetPhysicalValue(original) +
- // 'micros_to_add' and which retains the same logical value.
+ // 'to_add' and which retains the same logical value.
static Timestamp AddPhysicalTimeToTimestamp(const Timestamp& original,
const MonoDelta& to_add);
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/CMakeLists.txt b/src/kudu/tablet/CMakeLists.txt
index 9e8a00f..8fda82c 100644
--- a/src/kudu/tablet/CMakeLists.txt
+++ b/src/kudu/tablet/CMakeLists.txt
@@ -115,6 +115,7 @@ ADD_KUDU_TEST(delta_compaction-test)
ADD_KUDU_TEST(mt-rowset_delta_compaction-test)
ADD_KUDU_TEST(major_delta_compaction-test)
ADD_KUDU_TEST(transactions/transaction_tracker-test)
+ADD_KUDU_TEST(tablet_history_gc-test)
ADD_KUDU_TEST(tablet_peer-test)
ADD_KUDU_TEST(tablet_random_access-test)
ADD_KUDU_TEST(tablet_throttle-test)
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/compaction-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction-test.cc b/src/kudu/tablet/compaction-test.cc
index 738c8ce..923f3ea 100644
--- a/src/kudu/tablet/compaction-test.cc
+++ b/src/kudu/tablet/compaction-test.cc
@@ -208,7 +208,7 @@ class TestCompaction : public KuduRowSetTest {
BloomFilterSizing::BySizeAndFPRate(32*1024, 0.01f),
roll_threshold);
ASSERT_OK(rsw.Open());
- ASSERT_OK(FlushCompactionInput(input, snap, &rsw));
+ ASSERT_OK(FlushCompactionInput(input, snap, HistoryGcOpts::Disabled(), &rsw));
ASSERT_OK(rsw.Finish());
vector<shared_ptr<RowSetMetadata> > metas;
@@ -378,7 +378,8 @@ class TestCompaction : public KuduRowSetTest {
BloomFilterSizing::BySizeAndFPRate(32 * 1024, 0.01f),
1024 * 1024); // 1 MB
ASSERT_OK(rdrsw.Open());
- ASSERT_OK(FlushCompactionInput(compact_input.get(), merge_snap, &rdrsw));
+ ASSERT_OK(FlushCompactionInput(compact_input.get(), merge_snap, HistoryGcOpts::Disabled(),
+ &rdrsw));
ASSERT_OK(rdrsw.Finish());
}
}
@@ -575,7 +576,8 @@ TEST_F(TestCompaction, TestOneToOne) {
string dummy_name = "";
- ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), snap, snap2, { rs }));
+ ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), HistoryGcOpts::Disabled(), snap, snap2,
+ { rs }));
// If we look at the contents of the DiskRowSet now, we should see the "re-updated" data.
vector<string> out;
@@ -628,10 +630,10 @@ TEST_F(TestCompaction, TestKUDU102) {
string dummy_name = "";
// This would fail without KUDU-102
- ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), snap, snap2, { rs, rs_b }));
+ ASSERT_OK(ReupdateMissedDeltas(dummy_name, input.get(), HistoryGcOpts::Disabled(), snap, snap2,
+ { rs, rs_b }));
}
-
// Test compacting when all of the inputs and the output have the same schema
TEST_F(TestCompaction, TestMerge) {
vector<Schema> schemas;
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.cc b/src/kudu/tablet/compaction.cc
index 37a8877..53bf505 100644
--- a/src/kudu/tablet/compaction.cc
+++ b/src/kudu/tablet/compaction.cc
@@ -30,6 +30,7 @@
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/server/hybrid_clock.h"
#include "kudu/tablet/cfile_set.h"
#include "kudu/tablet/delta_store.h"
#include "kudu/tablet/delta_tracker.h"
@@ -38,6 +39,7 @@
#include "kudu/tablet/transactions/write_transaction.h"
#include "kudu/util/debug/trace_event.h"
+using kudu::server::HybridClock;
using std::shared_ptr;
using std::unique_ptr;
using std::unordered_set;
@@ -593,9 +595,34 @@ void RowSetsInCompaction::DumpToLog() const {
}
}
+void RemoveAncientUndos(const HistoryGcOpts& history_gc_opts, CompactionInputRow* row) {
+ if (!history_gc_opts.gc_enabled()) {
+ 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;
+ 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;
+ }
+ break;
+ }
+ prev_undo = undo_mut;
+ undo_mut = undo_mut->next();
+ }
+}
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,
@@ -603,12 +630,14 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
RowBlockRow* dst_row,
bool* is_garbage_collected,
uint64_t* num_rows_history_truncated) {
- // TODO actually perform garbage collection (KUDU-236).
- // Right now we persist all mutations.
+ // In most cases, we don't want to remove all traces of this row after
+ // 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;
#define ERROR_LOG_CONTEXT \
@@ -628,6 +657,8 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
Mutation* undo_head = const_cast<Mutation*>(src_row.undo_head);
Mutation* redo_head = 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).
for (const Mutation *redo_mut = src_row.redo_head;
redo_mut != nullptr;
redo_mut = redo_mut->acquire_next()) {
@@ -667,6 +698,11 @@ 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());
@@ -685,6 +721,8 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
redo_decoder.TwiddleDeleteStatus(&is_deleted);
if (redo_decoder.is_reinsert()) {
+ *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.
@@ -708,22 +746,30 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
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";
+ LOG(WARNING) << "Found REINSERT REDO truncating row history for " << ERROR_LOG_CONTEXT
+ << " Note: this warning will appear only for the first truncated row";
}
if (PREDICT_FALSE(VLOG_IS_ON(2))) {
VLOG(2) << "Found REINSERT REDO, cannot create UNDO for it, resetting row history "
- " under snap: " << snap.ToString() << ERROR_LOG_CONTEXT;
+ "under snapshot: " << snap.ToString() << ERROR_LOG_CONTEXT;
}
+
+ // Is a delete.
} else {
- // Delete mutations are left as redos
+ // Delete mutations are left as redos. Encode the DELETE as a redo.
undo_encoder.SetToDelete();
- // Encode the DELETE as a redo
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;
+ }
+
}
} else {
LOG(FATAL) << "Unknown mutation type!" << ERROR_LOG_CONTEXT;
@@ -740,6 +786,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
Status FlushCompactionInput(CompactionInput* input,
const MvccSnapshot& snap,
+ const HistoryGcOpts& history_gc_opts,
RollingDiskRowSetWriter* out) {
RETURN_NOT_OK(input->Init());
vector<CompactionInputRow> rows;
@@ -754,28 +801,32 @@ Status FlushCompactionInput(CompactionInput* input,
RETURN_NOT_OK(input->PrepareBlock(&rows));
int n = 0;
- for (const CompactionInputRow &input_row : rows) {
+ for (int i = 0; i < rows.size(); i++) {
+ CompactionInputRow* input_row = &rows[i];
RETURN_NOT_OK(out->RollIfNecessary());
- const Schema* schema = input_row.row.schema();
+ const Schema* schema = input_row->row.schema();
DCHECK_SCHEMA_EQ(*schema, out->schema());
DCHECK(schema->has_column_ids());
RowBlockRow dst_row = block.row(n);
- RETURN_NOT_OK(CopyRow(input_row.row, &dst_row, static_cast<Arena*>(nullptr)));
+ RETURN_NOT_OK(CopyRow(input_row->row, &dst_row, static_cast<Arena*>(nullptr)));
- DVLOG(2) << "Input Row: " << dst_row.schema()->DebugRow(dst_row) <<
- " RowId: " << input_row.row.row_index() <<
- " Undo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.undo_head) <<
- " Redo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.redo_head);
+ DVLOG(3) << "Input Row: " << dst_row.schema()->DebugRow(dst_row) <<
+ " RowIndex: " << input_row->row.row_index() <<
+ " Undo Mutations: " << Mutation::StringifyMutationList(*schema, input_row->undo_head) <<
+ " Redo Mutations: " << Mutation::StringifyMutationList(*schema, input_row->redo_head);
// Collect the new UNDO/REDO mutations.
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,
+ *input_row,
+ history_gc_opts,
schema,
&new_undos_head,
&new_redos_head,
@@ -784,33 +835,32 @@ Status FlushCompactionInput(CompactionInput* input,
&is_garbage_collected,
&num_rows_history_truncated));
+ DVLOG(4) << "Output Row: " << dst_row.schema()->DebugRow(dst_row) <<
+ "; RowIndex: " << dst_row.row_index() <<
+ "; Undo Mutations: " << Mutation::StringifyMutationList(*schema, new_undos_head) <<
+ "; Redo Mutations: " << Mutation::StringifyMutationList(*schema, new_redos_head) <<
+ "; Was garbage collected? " << is_garbage_collected <<
+ "; Was history truncated? " <<
+ (num_rows_history_truncated > prev_num_rows_history_truncated);
+
// Whether this row was garbage collected
if (is_garbage_collected) {
- DVLOG(2) << "Garbage Collected!";
// Don't flush the row.
continue;
}
- rowid_t index_in_current_drs_;
+ rowid_t index_in_current_drs;
- // We should always have UNDO deltas, until we implement delta GC. For now,
- // this is a convenient assertion to catch bugs like KUDU-632.
- CHECK(new_undos_head != nullptr) <<
- "Writing an output row with no UNDOs: "
- "Input Row: " << dst_row.schema()->DebugRow(dst_row) <<
- " RowId: " << input_row.row.row_index() <<
- " Undo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.undo_head) <<
- " Redo Mutations: " << Mutation::StringifyMutationList(*schema, input_row.redo_head);
- out->AppendUndoDeltas(dst_row.row_index(), new_undos_head, &index_in_current_drs_);
+ if (new_undos_head != nullptr) {
+ out->AppendUndoDeltas(dst_row.row_index(), new_undos_head, &index_in_current_drs);
+ }
if (new_redos_head != nullptr) {
- out->AppendRedoDeltas(dst_row.row_index(), new_redos_head, &index_in_current_drs_);
+ out->AppendRedoDeltas(dst_row.row_index(), new_redos_head, &index_in_current_drs);
}
- DVLOG(2) << "Output Row: " << dst_row.schema()->DebugRow(dst_row) <<
- " RowId: " << index_in_current_drs_
- << " Undo Mutations: " << Mutation::StringifyMutationList(*schema, new_undos_head)
- << " Redo Mutations: " << Mutation::StringifyMutationList(*schema, new_redos_head);
+ DVLOG(5) << "Output Row: " << dst_row.schema()->DebugRow(dst_row)
+ << "; RowId: " << index_in_current_drs;
n++;
if (n == block.nrows()) {
@@ -829,20 +879,21 @@ Status FlushCompactionInput(CompactionInput* input,
if (num_rows_history_truncated > 0) {
LOG(WARNING) << "Total " << num_rows_history_truncated
- << " rows lost some history due to REINSERT after DELETE";
+ << " rows lost some history due to REINSERT after DELETE";
}
return Status::OK();
}
Status ReupdateMissedDeltas(const string &tablet_name,
CompactionInput *input,
+ const HistoryGcOpts& history_gc_opts,
const MvccSnapshot &snap_to_exclude,
const MvccSnapshot &snap_to_include,
const RowSetVector &output_rowsets) {
TRACE_EVENT0("tablet", "ReupdateMissedDeltas");
RETURN_NOT_OK(input->Init());
- VLOG(1) << "Re-updating missed deltas between snapshot " <<
+ VLOG(1) << "Reupdating missed deltas between snapshot " <<
snap_to_exclude.ToString() << " and " << snap_to_include.ToString();
// Collect the delta trackers that we'll push the updates into.
@@ -851,7 +902,7 @@ Status ReupdateMissedDeltas(const string &tablet_name,
delta_trackers.push_back(down_cast<DiskRowSet *>(rs.get())->delta_tracker());
}
- // The map of updated delta trackers, indexed by id.
+ // The set of updated delta trackers.
unordered_set<DeltaTracker*> updated_trackers;
// When we apply the updates to the new DMS, there is no need to anchor them
@@ -873,27 +924,39 @@ Status ReupdateMissedDeltas(const string &tablet_name,
RETURN_NOT_OK(key_projector.Init());
faststring buf;
- rowid_t row_idx = 0;
+ rowid_t output_row_offset = 0;
while (input->HasMoreBlocks()) {
RETURN_NOT_OK(input->PrepareBlock(&rows));
for (const CompactionInputRow &row : rows) {
- DVLOG(2) << "Revisiting row: " << schema->DebugRow(row.row) <<
+ DVLOG(4) << "Revisiting row: " << schema->DebugRow(row.row) <<
" Redo Mutations: " << Mutation::StringifyMutationList(*schema, row.redo_head) <<
" Undo Mutations: " << Mutation::StringifyMutationList(*schema, row.undo_head);
+ bool is_garbage_collected = false;
for (const Mutation *mut = row.redo_head;
mut != nullptr;
mut = mut->acquire_next()) {
+ is_garbage_collected = false;
RowChangeListDecoder decoder(mut->changelist());
RETURN_NOT_OK(decoder.Init());
if (snap_to_exclude.IsCommitted(mut->timestamp())) {
// This update was already taken into account in the first phase of the
- // compaction.
+ // compaction. We don't need to reapply it.
+
+ // But was this row GCed at flush time?
+ if (decoder.is_delete() &&
+ history_gc_opts.IsAncientHistory(mut->timestamp())) {
+ is_garbage_collected = true;
+ }
continue;
}
+ // We should never be able to get to this point if a row has been
+ // garbage collected.
+ DCHECK(!is_garbage_collected);
+
// We should never see a REINSERT in an input RowSet which was not
// caught in the original flush. REINSERT only occurs when an INSERT is
// done to a row when a ghost is already present for that row in
@@ -920,7 +983,7 @@ Status ReupdateMissedDeltas(const string &tablet_name,
// Therefore, it's already present in the output rowset, and we don't need
// to copy it in.
- DVLOG(2) << "Skipping already-duplicated delta for row " << row_idx
+ DVLOG(3) << "Skipping already-duplicated delta for row " << output_row_offset
<< " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
continue;
}
@@ -928,15 +991,15 @@ Status ReupdateMissedDeltas(const string &tablet_name,
// Otherwise, this is an update that arrived after the snapshot for the first
// pass, but before the DuplicatingRowSet was swapped in. We need to transfer
// this over to the output rowset.
- DVLOG(1) << "Flushing missed delta for row " << row_idx
- << " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
+ DVLOG(3) << "Flushing missed delta for row " << output_row_offset
+ << " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
DeltaTracker *cur_tracker = delta_trackers.front();
// The index on the input side isn't necessarily the index on the output side:
// we may have output several small DiskRowSets, so we need to find the index
// relative to the current one.
- int64_t idx_in_delta_tracker = row_idx - delta_tracker_base_row;
+ int64_t idx_in_delta_tracker = output_row_offset - delta_tracker_base_row;
while (idx_in_delta_tracker >= cur_tracker->num_rows()) {
// If the current index is higher than the total number of rows in the current
// DeltaTracker, that means we're now processing the next one in the list.
@@ -955,7 +1018,7 @@ Status ReupdateMissedDeltas(const string &tablet_name,
mut->changelist(),
max_op_id,
result.get());
- DCHECK(s.ok()) << "Failed update on compaction for row " << row_idx
+ DCHECK(s.ok()) << "Failed update on compaction for row " << output_row_offset
<< " @" << mut->timestamp() << ": " << mut->changelist().ToString(*schema);
if (s.ok()) {
// Update the set of delta trackers with the one we've just updated.
@@ -963,15 +1026,19 @@ Status ReupdateMissedDeltas(const string &tablet_name,
}
}
- // TODO when garbage collection kicks in we need to take care that
- // CGed rows do not increment this.
- row_idx++;
+ if (is_garbage_collected) {
+ DVLOG(4) << "Skipping GCed input row: " << schema->DebugRow(row.row)
+ << " while reupdating missed deltas";
+ } else {
+ // Important: We must ensure that GCed rows do not increment the output
+ // row offset.
+ output_row_offset++;
+ }
}
RETURN_NOT_OK(input->FinishBlock());
}
-
// Flush the trackers that got updated, this will make sure that all missed deltas
// get flushed before we update the tablet's metadata at the end of compaction/flush.
// Note that we don't flush the metadata here, as to we will update the metadata
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/compaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/compaction.h b/src/kudu/tablet/compaction.h
index 9258ea0..b706879 100644
--- a/src/kudu/tablet/compaction.h
+++ b/src/kudu/tablet/compaction.h
@@ -32,6 +32,49 @@ namespace tablet {
struct CompactionInputRow;
class WriteTransactionState;
+// Options related to tablet history garbage collection.
+class HistoryGcOpts {
+ public:
+ static HistoryGcOpts Enabled(Timestamp ahm) {
+ return HistoryGcOpts(true, ahm);
+ }
+
+ static HistoryGcOpts Disabled() {
+ return HistoryGcOpts(false, Timestamp(0));
+ }
+
+ // Returns true if Timestamp 't' is considered "ancient history" and is
+ // eligible for garbage collection. If GC is disabled, will return false for
+ // any Timestamp with a value >= 0 (that is, all valid Timestamps).
+ bool IsAncientHistory(Timestamp t) const {
+ return t.ComesBefore(ancient_history_mark_);
+ }
+
+ // Returns true if history GC is enabled.
+ bool gc_enabled() const {
+ return gc_enabled_;
+ }
+
+ // Returns the ancient history mark.
+ Timestamp ancient_history_mark() const {
+ return ancient_history_mark_;
+ }
+
+ private:
+ HistoryGcOpts(bool gc_enabled, Timestamp ahm)
+ : gc_enabled_(gc_enabled),
+ ancient_history_mark_(ahm) {
+ }
+
+ // Whether historical records prior to the ancient history mark should be
+ // garbage-collected (deleted).
+ const bool gc_enabled_;
+
+ // A timestamp prior to which no history will be preserved.
+ // Ignored if 'enabled' != GC_ENABLED.
+ const Timestamp ancient_history_mark_;
+};
+
// Interface for an input feeding into a compaction or flush.
class CompactionInput {
public:
@@ -110,6 +153,8 @@ class RowSetsInCompaction {
};
// One row yielded by CompactionInput::PrepareBlock.
+// Looks like this (assuming n UNDO records and m REDO records):
+// UNDO_n <- ... <- UNDO_1 <- UNDO_head <- row -> REDO_head -> REDO_1 -> ... -> REDO_m
struct CompactionInputRow {
// The compaction input base row.
RowBlockRow row;
@@ -119,6 +164,11 @@ struct CompactionInputRow {
Mutation* undo_head;
};
+// 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);
+
// 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
// handling depends on the nature of the operation being performed:
@@ -128,9 +178,20 @@ struct CompactionInputRow {
// belonging to 'dst_row'. Those that don't belong to that schema are
// ignored.
//
-// Currently, 'is_garbage_collected' is always false (KUDU-236).
+// '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,
@@ -139,7 +200,6 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
bool* is_garbage_collected,
uint64_t* num_rows_history_truncated);
-
// 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.
//
@@ -147,6 +207,7 @@ Status ApplyMutationsAndGenerateUndos(const MvccSnapshot& snap,
// no longer be useful.
Status FlushCompactionInput(CompactionInput *input,
const MvccSnapshot &snap,
+ const HistoryGcOpts& history_gc_opts,
RollingDiskRowSetWriter *out);
// Iterate through this compaction input, finding any mutations which came between
@@ -161,6 +222,7 @@ Status FlushCompactionInput(CompactionInput *input,
// yield no further rows.
Status ReupdateMissedDeltas(const string &tablet_name,
CompactionInput *input,
+ const HistoryGcOpts& history_gc_opts,
const MvccSnapshot &snap_to_exclude,
const MvccSnapshot &snap_to_include,
const RowSetVector &output_rowsets);
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/delta_compaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_compaction.cc b/src/kudu/tablet/delta_compaction.cc
index 62bca8e..59a6da2 100644
--- a/src/kudu/tablet/delta_compaction.cc
+++ b/src/kudu/tablet/delta_compaction.cc
@@ -61,17 +61,19 @@ MajorDeltaCompaction::MajorDeltaCompaction(
FsManager* fs_manager, const Schema& base_schema, CFileSet* base_data,
unique_ptr<DeltaIterator> delta_iter,
vector<shared_ptr<DeltaStore> > included_stores,
- const vector<ColumnId>& col_ids)
+ vector<ColumnId> col_ids,
+ HistoryGcOpts history_gc_opts)
: fs_manager_(fs_manager),
base_schema_(base_schema),
- column_ids_(col_ids),
+ column_ids_(std::move(col_ids)),
+ history_gc_opts_(std::move(history_gc_opts)),
base_data_(base_data),
included_stores_(std::move(included_stores)),
delta_iter_(std::move(delta_iter)),
redo_delta_mutations_written_(0),
undo_delta_mutations_written_(0),
state_(kInitialized) {
- CHECK(!col_ids.empty());
+ CHECK(!column_ids_.empty());
}
MajorDeltaCompaction::~MajorDeltaCompaction() {
@@ -132,24 +134,32 @@ Status MajorDeltaCompaction::FlushRowSetAndDeltas() {
vector<CompactionInputRow> input_rows;
input_rows.resize(block.nrows());
for (int i = 0; i < block.nrows(); i++) {
- CompactionInputRow &input_row = input_rows.at(i);
- input_row.row.Reset(&block, i);
- input_row.redo_head = redo_mutation_block[i];
- Mutation::ReverseMutationList(&input_row.redo_head);
- input_row.undo_head = nullptr;
+ CompactionInputRow* input_row = &input_rows[i];
+ input_row->row.Reset(&block, i);
+ input_row->redo_head = redo_mutation_block[i];
+ Mutation::ReverseMutationList(&input_row->redo_head);
+ input_row->undo_head = nullptr;
RowBlockRow dst_row = block.row(i);
- RETURN_NOT_OK(CopyRow(input_row.row, &dst_row, static_cast<Arena*>(nullptr)));
+ RETURN_NOT_OK(CopyRow(input_row->row, &dst_row, static_cast<Arena*>(nullptr)));
Mutation* new_undos_head = nullptr;
// We're ignoring the result from new_redos_head because we'll find them
// later at step 5.
Mutation* new_redos_head = nullptr;
+ DVLOG(3) << "MDC: Input Row: " << dst_row.schema()->DebugRow(dst_row)
+ << " RowIndex: " << input_row->row.row_index()
+ << " Undo Mutations: (not shown)"
+ << " Redo Mutations: "
+ << Mutation::StringifyMutationList(base_schema_, input_row->redo_head);
+
bool is_garbage_collected;
+ RemoveAncientUndos(history_gc_opts_, input_row);
RETURN_NOT_OK(ApplyMutationsAndGenerateUndos(snap,
- input_row,
+ *input_row,
+ history_gc_opts_,
&base_schema_,
&new_undos_head,
&new_redos_head,
@@ -158,9 +168,12 @@ Status MajorDeltaCompaction::FlushRowSetAndDeltas() {
&is_garbage_collected,
&num_rows_history_truncated));
- VLOG(2) << "Output Row: " << dst_row.schema()->DebugRow(dst_row)
- << " Undo Mutations: " << Mutation::StringifyMutationList(partial_schema_, new_undos_head)
- << " Redo Mutations: " << Mutation::StringifyMutationList(partial_schema_, new_redos_head);
+ DVLOG(3) << "MDC: Output Row: " << dst_row.schema()->DebugRow(dst_row)
+ << " RowIndex: " << input_row->row.row_index()
+ << " Generated Undo Mutations: "
+ << Mutation::StringifyMutationList(partial_schema_, new_undos_head)
+ << " Updated Redo Mutations: "
+ << Mutation::StringifyMutationList(partial_schema_, new_redos_head);
// We only create a new undo delta file if we need to.
if (new_undos_head != nullptr && !new_undo_delta_writer_) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/delta_compaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/delta_compaction.h b/src/kudu/tablet/delta_compaction.h
index d797072..85760bb 100644
--- a/src/kudu/tablet/delta_compaction.h
+++ b/src/kudu/tablet/delta_compaction.h
@@ -24,6 +24,7 @@
#include <vector>
#include "kudu/cfile/cfile_writer.h"
+#include "kudu/tablet/compaction.h"
#include "kudu/tablet/deltafile.h"
namespace kudu {
@@ -56,7 +57,8 @@ class MajorDeltaCompaction {
FsManager* fs_manager, const Schema& base_schema, CFileSet* base_data,
std::unique_ptr<DeltaIterator> delta_iter,
std::vector<std::shared_ptr<DeltaStore> > included_stores,
- const std::vector<ColumnId>& col_ids);
+ std::vector<ColumnId> col_ids,
+ HistoryGcOpts history_gc_opts);
~MajorDeltaCompaction();
// Executes the compaction.
@@ -103,6 +105,8 @@ class MajorDeltaCompaction {
// The column ids to compact.
const std::vector<ColumnId> column_ids_;
+ const HistoryGcOpts history_gc_opts_;
+
// Inputs:
//-----------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/diskrowset.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/diskrowset.cc b/src/kudu/tablet/diskrowset.cc
index 5feeab4..e7fabf4 100644
--- a/src/kudu/tablet/diskrowset.cc
+++ b/src/kudu/tablet/diskrowset.cc
@@ -507,7 +507,7 @@ Status DiskRowSet::MinorCompactDeltaStores() {
return delta_tracker_->Compact();
}
-Status DiskRowSet::MajorCompactDeltaStores() {
+Status DiskRowSet::MajorCompactDeltaStores(HistoryGcOpts history_gc_opts) {
vector<ColumnId> col_ids;
delta_tracker_->GetColumnIdsWithUpdates(&col_ids);
@@ -515,16 +515,17 @@ Status DiskRowSet::MajorCompactDeltaStores() {
return Status::OK();
}
- return MajorCompactDeltaStoresWithColumnIds(col_ids);
+ return MajorCompactDeltaStoresWithColumnIds(col_ids, std::move(history_gc_opts));
}
-Status DiskRowSet::MajorCompactDeltaStoresWithColumnIds(const vector<ColumnId>& col_ids) {
+Status DiskRowSet::MajorCompactDeltaStoresWithColumnIds(const vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts) {
TRACE_EVENT0("tablet", "DiskRowSet::MajorCompactDeltaStores");
std::lock_guard<Mutex> l(*delta_tracker()->compact_flush_lock());
// TODO: do we need to lock schema or anything here?
gscoped_ptr<MajorDeltaCompaction> compaction;
- RETURN_NOT_OK(NewMajorDeltaCompaction(col_ids, &compaction));
+ RETURN_NOT_OK(NewMajorDeltaCompaction(col_ids, std::move(history_gc_opts), &compaction));
RETURN_NOT_OK(compaction->Compact());
@@ -547,6 +548,7 @@ Status DiskRowSet::MajorCompactDeltaStoresWithColumnIds(const vector<ColumnId>&
}
Status DiskRowSet::NewMajorDeltaCompaction(const vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts,
gscoped_ptr<MajorDeltaCompaction>* out) const {
DCHECK(open_);
shared_lock<rw_spinlock> l(component_lock_.get_lock());
@@ -566,8 +568,9 @@ Status DiskRowSet::NewMajorDeltaCompaction(const vector<ColumnId>& col_ids,
*schema,
base_data_.get(),
std::move(delta_iter),
- included_stores,
- col_ids));
+ std::move(included_stores),
+ col_ids,
+ std::move(history_gc_opts)));
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/diskrowset.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/diskrowset.h b/src/kudu/tablet/diskrowset.h
index d638336..d096793 100644
--- a/src/kudu/tablet/diskrowset.h
+++ b/src/kudu/tablet/diskrowset.h
@@ -61,6 +61,7 @@ class CFileSet;
class DeltaFileWriter;
class DeltaStats;
class DeltaTracker;
+class HistoryGcOpts;
class MultiColumnWriter;
class Mutation;
class OperationResultPB;
@@ -346,7 +347,7 @@ class DiskRowSet : public RowSet {
double DeltaStoresCompactionPerfImprovementScore(DeltaCompactionType type) const OVERRIDE;
// Major compacts all the delta files for all the columns.
- Status MajorCompactDeltaStores();
+ Status MajorCompactDeltaStores(HistoryGcOpts history_gc_opts);
std::mutex *compact_flush_lock() OVERRIDE {
return &compact_flush_lock_;
@@ -370,6 +371,7 @@ class DiskRowSet : public RowSet {
FRIEND_TEST(TestRowSet, TestRowSetUpdate);
FRIEND_TEST(TestRowSet, TestDMSFlush);
FRIEND_TEST(TestCompaction, TestOneToOne);
+ FRIEND_TEST(TabletHistoryGcTest, TestMajorDeltaCompactionOnSubsetOfColumns);
friend class CompactionInput;
friend class Tablet;
@@ -382,10 +384,12 @@ class DiskRowSet : public RowSet {
// Create a new major delta compaction object to compact the specified columns.
Status NewMajorDeltaCompaction(const std::vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts,
gscoped_ptr<MajorDeltaCompaction>* out) const;
// Major compacts all the delta files for the specified columns.
- Status MajorCompactDeltaStoresWithColumnIds(const std::vector<ColumnId>& col_ids);
+ Status MajorCompactDeltaStoresWithColumnIds(const std::vector<ColumnId>& col_ids,
+ HistoryGcOpts history_gc_opts);
std::shared_ptr<RowSetMetadata> rowset_metadata_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/mutation.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/mutation.h b/src/kudu/tablet/mutation.h
index 1eb73ca..bb3e464 100644
--- a/src/kudu/tablet/mutation.h
+++ b/src/kudu/tablet/mutation.h
@@ -51,6 +51,8 @@ class Mutation {
}
Timestamp timestamp() const { return timestamp_; }
+
+ Mutation *next() { return next_; }
const Mutation *next() const { return next_; }
// Same as 'next()' but loads with 'Acquire' ordering semantics.
@@ -91,8 +93,6 @@ class Mutation {
template<bool ATOMIC>
void DoAppendToList(Mutation **list);
- DISALLOW_COPY_AND_ASSIGN(Mutation);
-
// The transaction ID which made this mutation. If this transaction is not
// committed in the snapshot of the reader, this mutation should be ignored.
Timestamp timestamp_;
@@ -104,6 +104,8 @@ class Mutation {
// The actual encoded RowChangeList
char changelist_data_[0];
+
+ DISALLOW_COPY_AND_ASSIGN(Mutation);
};
template<class ArenaType>
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/rowset_metadata.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/rowset_metadata.h b/src/kudu/tablet/rowset_metadata.h
index 14ed190..c195d8f 100644
--- a/src/kudu/tablet/rowset_metadata.h
+++ b/src/kudu/tablet/rowset_metadata.h
@@ -43,7 +43,7 @@ namespace tablet {
class RowSetMetadataUpdate;
class TabletMetadata;
-// Keeps tracks of the RowSet data blocks.
+// Keeps track of the RowSet data blocks.
//
// On each tablet MemRowSet flush, a new RowSetMetadata is created,
// and the DiskRowSetWriter will create and write the "immutable" blocks for
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-harness.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-harness.h b/src/kudu/tablet/tablet-harness.h
index f8f376c..fd09c5a 100644
--- a/src/kudu/tablet/tablet-harness.h
+++ b/src/kudu/tablet/tablet-harness.h
@@ -25,6 +25,7 @@
#include "kudu/common/schema.h"
#include "kudu/consensus/log_anchor_registry.h"
#include "kudu/consensus/metadata.pb.h"
+#include "kudu/server/hybrid_clock.h"
#include "kudu/server/logical_clock.h"
#include "kudu/tablet/tablet.h"
#include "kudu/util/env.h"
@@ -60,16 +61,22 @@ static std::pair<PartitionSchema, Partition> CreateDefaultPartition(const Schema
class TabletHarness {
public:
struct Options {
+ enum ClockType {
+ HYBRID_CLOCK,
+ LOGICAL_CLOCK
+ };
explicit Options(string root_dir)
: env(Env::Default()),
tablet_id("test_tablet_id"),
root_dir(std::move(root_dir)),
- enable_metrics(true) {}
+ enable_metrics(true),
+ clock_type(LOGICAL_CLOCK) {}
Env* env;
string tablet_id;
string root_dir;
bool enable_metrics;
+ ClockType clock_type;
};
TabletHarness(const Schema& schema, Options options)
@@ -99,7 +106,12 @@ class TabletHarness {
metrics_registry_.reset(new MetricRegistry());
}
- clock_ = server::LogicalClock::CreateStartingAt(Timestamp::kInitialTimestamp);
+ if (options_.clock_type == Options::LOGICAL_CLOCK) {
+ clock_.reset(server::LogicalClock::CreateStartingAt(Timestamp::kInitialTimestamp));
+ } else {
+ clock_.reset(new server::HybridClock());
+ RETURN_NOT_OK(clock_->Init());
+ }
tablet_.reset(new Tablet(metadata,
clock_,
std::shared_ptr<MemTracker>(),
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-test-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test-base.h b/src/kudu/tablet/tablet-test-base.h
index 78d2916..a00c62b 100644
--- a/src/kudu/tablet/tablet-test-base.h
+++ b/src/kudu/tablet/tablet-test-base.h
@@ -18,6 +18,7 @@
#define KUDU_TABLET_TABLET_TEST_BASE_H
#include <algorithm>
+#include <boost/optional.hpp>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <limits>
@@ -304,8 +305,11 @@ typedef ::testing::Types<
template<class TESTSETUP>
class TabletTestBase : public KuduTabletTest {
public:
- TabletTestBase() :
- KuduTabletTest(TESTSETUP::CreateSchema()),
+ typedef std::function<bool(int32_t, int32_t)> TestRowVerifier;
+
+ TabletTestBase(TabletHarness::Options::ClockType clock_type =
+ TabletHarness::Options::ClockType::LOGICAL_CLOCK) :
+ KuduTabletTest(TESTSETUP::CreateSchema(), clock_type),
setup_(),
max_rows_(setup_.GetMaxRows()),
arena_(1024, 4*1024*1024)
@@ -398,59 +402,90 @@ class TabletTestBase : public KuduTabletTest {
}
template <class RowType>
- void VerifyRow(const RowType& row, int64_t key_idx, int32_t val) {
+ void VerifyRow(const RowType& row, int32_t key_idx, int32_t val) {
ASSERT_EQ(setup_.FormatDebugRow(key_idx, val, false), schema_.DebugRow(row));
}
- void VerifyTestRows(int64_t first_row, uint64_t expected_count) {
+ void VerifyTestRows(int32_t first_row, uint64_t expected_count) {
+ VerifyTestRowsWithVerifier(first_row, expected_count, boost::none);
+ }
+
+ void VerifyTestRowsWithVerifier(int32_t first_row, uint64_t expected_count,
+ boost::optional<TestRowVerifier> verifier) {
gscoped_ptr<RowwiseIterator> iter;
ASSERT_OK(tablet()->NewRowIterator(client_schema_, &iter));
+ VerifyTestRowsWithRowIteratorAndVerifier(first_row, expected_count, std::move(iter), verifier);
+ }
+
+ void VerifyTestRowsWithTimestampAndVerifier(int32_t first_row, uint64_t expected_count,
+ Timestamp timestamp,
+ boost::optional<TestRowVerifier> verifier) {
+ gscoped_ptr<RowwiseIterator> iter;
+ ASSERT_OK(tablet()->NewRowIterator(client_schema_, MvccSnapshot(timestamp), Tablet::UNORDERED,
+ &iter));
+ VerifyTestRowsWithRowIteratorAndVerifier(first_row, expected_count, std::move(iter), verifier);
+ }
+
+ void VerifyTestRowsWithRowIteratorAndVerifier(int32_t first_row, uint64_t expected_row_count,
+ gscoped_ptr<RowwiseIterator> iter,
+ boost::optional<TestRowVerifier> verifier) {
ASSERT_OK(iter->Init(NULL));
- int batch_size = std::max(
- (size_t)1, std::min((size_t)(expected_count / 10),
- 4*1024*1024 / schema_.byte_size()));
+ int batch_size = std::max<size_t>(1, std::min<size_t>(expected_row_count / 10,
+ 4L * 1024 * 1024 / schema_.byte_size()));
Arena arena(32*1024, 256*1024);
RowBlock block(schema_, batch_size, &arena);
- if (expected_count > INT_MAX) {
- LOG(INFO) << "Not checking rows for duplicates -- duplicates expected since "
- << "there were more than " << INT_MAX << " rows inserted.";
- return;
+ bool check_for_dups = true;
+ if (expected_row_count > INT_MAX) {
+ check_for_dups = false;
+ LOG(WARNING) << "Not checking rows for duplicates -- duplicates expected since "
+ << "there were more than " << INT_MAX << " rows inserted.";
}
// Keep a bitmap of which rows have been seen from the requested
// range.
std::vector<bool> seen_rows;
- seen_rows.resize(expected_count);
+ seen_rows.resize(expected_row_count);
+ uint64_t actual_row_count = 0;
while (iter->HasNext()) {
ASSERT_OK_FAST(iter->NextBlock(&block));
RowBlockRow rb_row = block.row(0);
if (VLOG_IS_ON(2)) {
VLOG(2) << "Fetched batch of " << block.nrows() << "\n"
- << "First row: " << schema_.DebugRow(rb_row);
+ << "First row: " << schema_.DebugRow(rb_row);
}
for (int i = 0; i < block.nrows(); i++) {
rb_row.Reset(&block, i);
int32_t key_idx = *schema_.ExtractColumnFromRow<INT32>(rb_row, 1);
- if (key_idx >= first_row && key_idx < first_row + expected_count) {
- size_t rel_idx = key_idx - first_row;
- if (seen_rows[rel_idx]) {
- FAIL() << "Saw row " << key_idx << " twice!\n"
- << "Row: " << schema_.DebugRow(rb_row);
+ if (key_idx >= first_row && block.selection_vector()->IsRowSelected(i)) {
+ actual_row_count++;
+ if (key_idx < first_row + expected_row_count) {
+ size_t rel_idx = key_idx - first_row;
+ if (check_for_dups && seen_rows[rel_idx]) {
+ FAIL() << "Saw row " << key_idx << " twice!\n"
+ << "Row: " << schema_.DebugRow(rb_row);
+ }
+ seen_rows[rel_idx] = true;
+ if (verifier) {
+ int32_t val = *schema_.ExtractColumnFromRow<INT32>(rb_row, 2);
+ ASSERT_TRUE((*verifier)(key_idx, val))
+ << "Key index: " << key_idx << ", value: " << val;
+ }
}
- seen_rows[rel_idx] = true;
}
}
}
// Verify that all the rows were seen.
- for (int i = 0; i < expected_count; i++) {
- ASSERT_EQ(true, seen_rows[i]) << "Never saw row: " << (i + first_row);
+ for (int i = 0; i < expected_row_count; i++) {
+ ASSERT_EQ(true, seen_rows[i]) << "Never saw row " << (i + first_row);
}
- LOG(INFO) << "Successfully verified " << expected_count << "rows";
+ ASSERT_EQ(expected_row_count, actual_row_count)
+ << "Expected row count didn't match actual row count";
+ LOG(INFO) << "Successfully verified " << expected_row_count << "rows";
}
// Iterate through the full table, stringifying the resulting rows
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/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 7cc8516..977987f 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -45,9 +45,12 @@ using std::vector;
class KuduTabletTest : public KuduTest {
public:
- explicit KuduTabletTest(const Schema& schema)
+ explicit KuduTabletTest(const Schema& schema,
+ TabletHarness::Options::ClockType clock_type =
+ TabletHarness::Options::ClockType::LOGICAL_CLOCK)
: schema_(schema.CopyWithColumnIds()),
- client_schema_(schema) {
+ client_schema_(schema),
+ clock_type_(clock_type) {
// Keep unit tests fast, but only if no one has set the flag explicitly.
if (google::GetCommandLineFlagInfoOrDie("enable_data_block_fsync").is_default) {
FLAGS_enable_data_block_fsync = false;
@@ -64,6 +67,7 @@ class KuduTabletTest : public KuduTest {
string dir = root_dir.empty() ? GetTestPath("fs_root") : root_dir;
TabletHarness::Options opts(dir);
opts.enable_metrics = true;
+ opts.clock_type = clock_type_;
bool first_time = harness_ == NULL;
harness_.reset(new TabletHarness(schema_, opts));
CHECK_OK(harness_->Create(first_time));
@@ -115,6 +119,7 @@ class KuduTabletTest : public KuduTest {
protected:
const Schema schema_;
const Schema client_schema_;
+ const TabletHarness::Options::ClockType clock_type_;
gscoped_ptr<TabletHarness> harness_;
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test.cc b/src/kudu/tablet/tablet-test.cc
index a8f97bc..f704f0d 100644
--- a/src/kudu/tablet/tablet-test.cc
+++ b/src/kudu/tablet/tablet-test.cc
@@ -553,17 +553,22 @@ TYPED_TEST(TestTablet, TestInsertsPersist) {
this->InsertTestRows(0, max_rows, 0);
ASSERT_EQ(max_rows, this->TabletCount());
+ // Get current timestamp.
+ Timestamp t = this->tablet()->clock()->Now();
+
// Flush it.
ASSERT_OK(this->tablet()->Flush());
ASSERT_EQ(max_rows, this->TabletCount());
- // Close and re-open tablet
+ // Close and re-open tablet.
+ // TODO: Should we be reopening the tablet in a different way to persist the
+ // clock / timestamps?
this->TabletReOpen();
// Ensure that rows exist
ASSERT_EQ(max_rows, this->TabletCount());
- this->VerifyTestRows(0, max_rows);
+ this->VerifyTestRowsWithTimestampAndVerifier(0, max_rows, t, boost::none);
// TODO: add some more data, re-flush
}
@@ -645,8 +650,6 @@ TYPED_TEST(TestTablet, TestMultipleUpdates) {
ASSERT_EQ(this->setup_.FormatDebugRow(1, 0, false), out_rows[1]);
}
-
-
TYPED_TEST(TestTablet, TestCompaction) {
uint64_t max_rows = this->ClampRowCount(FLAGS_testcompaction_num_rows);
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index faf1ae1..d223c1f 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -41,6 +41,7 @@
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/strings/numbers.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/server/hybrid_clock.h"
#include "kudu/tablet/compaction.h"
#include "kudu/tablet/compaction_policy.h"
#include "kudu/tablet/delta_compaction.h"
@@ -91,7 +92,6 @@ DEFINE_double(fault_crash_before_flush_tablet_meta_after_flush_mrs, 0.0,
"Fraction of the time, while flushing an MRS, to crash before flushing metadata");
TAG_FLAG(fault_crash_before_flush_tablet_meta_after_flush_mrs, unsafe);
-
DEFINE_int64(tablet_throttler_rpc_per_sec, 0,
"Maximum write RPC rate (op/s) allowed for a tablet, write RPC exceeding this "
"limit will be throttled. 0 means no limit.");
@@ -108,6 +108,12 @@ DEFINE_double(tablet_throttler_burst_factor, 1.0f,
"base rate.");
TAG_FLAG(tablet_throttler_burst_factor, experimental);
+DEFINE_int32(tablet_history_max_age_sec, -1,
+ "Number of seconds to retain tablet history. Reads initiated at a "
+ "snapshot that is older than this age will be rejected. "
+ "To disable history removal, set to -1.");
+TAG_FLAG(tablet_history_max_age_sec, advanced);
+
METRIC_DEFINE_entity(tablet);
METRIC_DEFINE_gauge_size(tablet, memrowset_size, "MemRowSet Memory Usage",
kudu::MetricUnit::kBytes,
@@ -116,21 +122,22 @@ METRIC_DEFINE_gauge_size(tablet, on_disk_size, "Tablet Size On Disk",
kudu::MetricUnit::kBytes,
"Size of this tablet on disk.");
+using base::subtle::Barrier_AtomicIncrement;
+using kudu::MaintenanceManager;
+using kudu::consensus::OpId;
+using kudu::consensus::MaximumOpId;
+using kudu::log::LogAnchorRegistry;
+using kudu::server::HybridClock;
using std::shared_ptr;
using std::string;
+using std::unique_ptr;
using std::unordered_set;
using std::vector;
+using strings::Substitute;
namespace kudu {
namespace tablet {
-using kudu::MaintenanceManager;
-using consensus::OpId;
-using consensus::MaximumOpId;
-using log::LogAnchorRegistry;
-using strings::Substitute;
-using base::subtle::Barrier_AtomicIncrement;
-
static CompactionPolicy *CreateCompactionPolicy() {
return new BudgetedCompactionPolicy(FLAGS_tablet_compaction_budget_mb);
}
@@ -683,10 +690,40 @@ Status Tablet::DoMajorDeltaCompaction(const vector<ColumnId>& col_ids,
shared_ptr<RowSet> input_rs) {
CHECK_EQ(state_, kOpen);
Status s = down_cast<DiskRowSet*>(input_rs.get())
- ->MajorCompactDeltaStoresWithColumnIds(col_ids);
+ ->MajorCompactDeltaStoresWithColumnIds(col_ids, GetHistoryGcOpts());
return s;
}
+bool Tablet::GetTabletAncientHistoryMark(Timestamp* ancient_history_mark) const {
+ // We currently only support history GC through a fully-instantiated tablet
+ // when using the HybridClock, since we can calculate the age of a mutation.
+ if (!clock_->HasPhysicalComponent() || FLAGS_tablet_history_max_age_sec < 0) {
+ return false;
+ }
+ Timestamp now = clock_->Now();
+ uint64_t now_micros = HybridClock::GetPhysicalValueMicros(now);
+ uint64_t max_age_micros = FLAGS_tablet_history_max_age_sec * 1000000ULL;
+ // Ensure that the AHM calculation doesn't underflow when
+ // '--tablet_history_max_age_sec' is set to a very high value.
+ if (max_age_micros <= now_micros) {
+ *ancient_history_mark =
+ HybridClock::TimestampFromMicrosecondsAndLogicalValue(
+ now_micros - max_age_micros,
+ HybridClock::GetLogicalValue(now));
+ } else {
+ *ancient_history_mark = Timestamp(0);
+ }
+ return true;
+}
+
+HistoryGcOpts Tablet::GetHistoryGcOpts() const {
+ Timestamp ancient_history_mark;
+ if (GetTabletAncientHistoryMark(&ancient_history_mark)) {
+ return HistoryGcOpts::Enabled(ancient_history_mark);
+ }
+ return HistoryGcOpts::Disabled();
+}
+
Status Tablet::Flush() {
TRACE_EVENT1("tablet", "Tablet::Flush", "id", tablet_id());
std::lock_guard<Semaphore> lock(rowsets_flush_sem_);
@@ -1255,7 +1292,9 @@ Status Tablet::DoCompactionOrFlush(const RowSetsInCompaction &input, int64_t mrs
RollingDiskRowSetWriter drsw(metadata_.get(), merge->schema(), bloom_sizing(),
compaction_policy_->target_rowset_size());
RETURN_NOT_OK_PREPEND(drsw.Open(), "Failed to open DiskRowSet for flush");
- RETURN_NOT_OK_PREPEND(FlushCompactionInput(merge.get(), flush_snap, &drsw),
+
+ HistoryGcOpts history_gc_opts = GetHistoryGcOpts();
+ RETURN_NOT_OK_PREPEND(FlushCompactionInput(merge.get(), flush_snap, history_gc_opts, &drsw),
"Flush to disk failed");
RETURN_NOT_OK_PREPEND(drsw.Finish(), "Failed to finish DRS writer");
@@ -1389,6 +1428,7 @@ Status Tablet::DoCompactionOrFlush(const RowSetsInCompaction &input, int64_t mrs
// rowsets is flushed.
RETURN_NOT_OK_PREPEND(ReupdateMissedDeltas(metadata_->tablet_id(),
merge.get(),
+ history_gc_opts,
flush_snap,
non_duplicated_txns_snap,
new_disk_rowsets),
@@ -1449,10 +1489,6 @@ Status Tablet::Compact(CompactFlags flags) {
// Step 1. Capture the rowsets to be merged
RETURN_NOT_OK_PREPEND(PickRowSetsToCompact(&input, flags),
"Failed to pick rowsets to compact");
- if (input.num_rowsets() < 2) {
- VLOG_WITH_PREFIX(1) << "Not enough rowsets to run compaction! Aborting...";
- return Status::OK();
- }
LOG_WITH_PREFIX(INFO) << "Compaction: stage 1 complete, picked "
<< input.num_rowsets() << " rowsets to compact";
if (compaction_hooks_) {
@@ -1725,6 +1761,7 @@ Status Tablet::FlushBiggestDMS() {
Status Tablet::CompactWorstDeltas(RowSet::DeltaCompactionType type) {
CHECK_EQ(state_, kOpen);
shared_ptr<RowSet> rs;
+
// We're required to grab the rowset's compact_flush_lock under the compact_select_lock_.
std::unique_lock<std::mutex> lock;
double perf_improv;
@@ -1734,12 +1771,11 @@ Status Tablet::CompactWorstDeltas(RowSet::DeltaCompactionType type) {
// under this lock.
std::lock_guard<std::mutex> compact_lock(compact_select_lock_);
perf_improv = GetPerfImprovementForBestDeltaCompactUnlocked(type, &rs);
- if (rs) {
- lock = std::unique_lock<std::mutex>(*rs->compact_flush_lock(), std::try_to_lock);
- CHECK(lock.owns_lock());
- } else {
+ if (!rs) {
return Status::OK();
}
+ lock = std::unique_lock<std::mutex>(*rs->compact_flush_lock(), std::try_to_lock);
+ CHECK(lock.owns_lock());
}
// We just released compact_select_lock_ so other compactions can select and run, but the
@@ -1749,14 +1785,14 @@ Status Tablet::CompactWorstDeltas(RowSet::DeltaCompactionType type) {
RETURN_NOT_OK_PREPEND(rs->MinorCompactDeltaStores(),
"Failed minor delta compaction on " + rs->ToString());
} else if (type == RowSet::MAJOR_DELTA_COMPACTION) {
- RETURN_NOT_OK_PREPEND(down_cast<DiskRowSet*>(rs.get())->MajorCompactDeltaStores(),
- "Failed major delta compaction on " + rs->ToString());
+ RETURN_NOT_OK_PREPEND(down_cast<DiskRowSet*>(rs.get())->MajorCompactDeltaStores(
+ GetHistoryGcOpts()), "Failed major delta compaction on " + rs->ToString());
}
return Status::OK();
}
double Tablet::GetPerfImprovementForBestDeltaCompact(RowSet::DeltaCompactionType type,
- shared_ptr<RowSet>* rs) const {
+ shared_ptr<RowSet>* rs) const {
std::lock_guard<std::mutex> compact_lock(compact_select_lock_);
return GetPerfImprovementForBestDeltaCompactUnlocked(type, rs);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/be719edc/src/kudu/tablet/tablet.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index 7663cde..f0b8b31 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -64,6 +64,7 @@ namespace tablet {
class AlterSchemaTransactionState;
class CompactionPolicy;
+class HistoryGcOpts;
class MemRowSet;
class MvccSnapshot;
struct RowOp;
@@ -346,6 +347,14 @@ class Tablet {
Status DoMajorDeltaCompaction(const std::vector<ColumnId>& column_ids,
std::shared_ptr<RowSet> input_rowset);
+ // Calculates the ancient history mark and returns true iff tablet history GC
+ // is enabled, which requires the use of a HybridClock.
+ // Otherwise, returns false.
+ bool GetTabletAncientHistoryMark(Timestamp* ancient_history_mark) const WARN_UNUSED_RESULT;
+
+ // Calculates history GC options based on properties of the Clock implementation.
+ HistoryGcOpts GetHistoryGcOpts() const;
+
// Method used by tests to retrieve all rowsets of this table. This
// will be removed once code for selecting the appropriate RowSet is
// finished and delta files is finished is part of Tablet class.
@@ -374,6 +383,8 @@ class Tablet {
// Return true if this RPC is allowed.
bool ShouldThrottleAllow(int64_t bytes);
+ scoped_refptr<server::Clock> clock() const { return clock_; }
+
static const char* kDMSMemTrackerId;
private:
friend class Iterator;