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/19 21:25:26 UTC
[1/3] kudu git commit: Python - Bump package version to 0.4.0
Repository: kudu
Updated Branches:
refs/heads/master 1f639da62 -> d87e3ff60
Python - Bump package version to 0.4.0
There have been additional changes to the python client since the
0.3.0 release so the version number needs bumped before the next
release.
Change-Id: Iff66be56a55f5f75c20bd2183d4ffad54edf46f1
Reviewed-on: http://gerrit.cloudera.org:8080/4444
Tested-by: Kudu Jenkins
Reviewed-by: David Ribeiro Alves <dr...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/4ba39e4d
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/4ba39e4d
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/4ba39e4d
Branch: refs/heads/master
Commit: 4ba39e4d2694b87eac0cc65f762ba0de293b72ad
Parents: 1f639da
Author: Jordan Birdsell <jo...@gmail.com>
Authored: Fri Sep 16 22:02:20 2016 -0400
Committer: Todd Lipcon <to...@apache.org>
Committed: Mon Sep 19 18:40:33 2016 +0000
----------------------------------------------------------------------
python/setup.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/4ba39e4d/python/setup.py
----------------------------------------------------------------------
diff --git a/python/setup.py b/python/setup.py
index 830a103..b3dcda4 100644
--- a/python/setup.py
+++ b/python/setup.py
@@ -32,7 +32,7 @@ if Cython.__version__ < '0.21.0':
raise Exception('Please upgrade to Cython 0.21.0 or newer')
MAJOR = 0
-MINOR = 3
+MINOR = 4
MICRO = 0
VERSION = '%d.%d.%d' % (MAJOR, MINOR, MICRO)
ISRELEASED = True
[3/3] kudu git commit: KUDU-1135 (part 1): avoid flushing cmeta to
disk twice when voting
Posted by to...@apache.org.
KUDU-1135 (part 1): avoid flushing cmeta to disk twice when voting
This adds a flag to the term advancement calls which prevents it from
flushing to disk. We use this flag during voting when we know we are
about to flush our vote to disk immediately following the term
advancement.
This is part 1: another patch will do the same optimization for the case
of starting an election.
Change-Id: Iecc55bc9e96dcdc918ede1190b7cbac719f95506
Reviewed-on: http://gerrit.cloudera.org:8080/4333
Reviewed-by: Mike Percy <mp...@apache.org>
Reviewed-by: David Ribeiro Alves <dr...@apache.org>
Tested-by: David Ribeiro Alves <dr...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/d87e3ff6
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/d87e3ff6
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/d87e3ff6
Branch: refs/heads/master
Commit: d87e3ff6036ba070710ecf383448bdfb8380378f
Parents: 9911c48
Author: Todd Lipcon <to...@apache.org>
Authored: Wed Sep 7 20:37:51 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Mon Sep 19 21:24:57 2016 +0000
----------------------------------------------------------------------
src/kudu/consensus/consensus_meta.cc | 5 +++-
src/kudu/consensus/consensus_meta.h | 7 ++++++
src/kudu/consensus/raft_consensus.cc | 26 ++++++++++++++------
src/kudu/consensus/raft_consensus.h | 7 ++++--
.../consensus/raft_consensus_quorum-test.cc | 26 ++++++++++++++++++++
src/kudu/consensus/raft_consensus_state.cc | 7 ++++--
src/kudu/consensus/raft_consensus_state.h | 16 +++++++++++-
7 files changed, 80 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/src/kudu/consensus/consensus_meta.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_meta.cc b/src/kudu/consensus/consensus_meta.cc
index 9426ceb..cdb4186 100644
--- a/src/kudu/consensus/consensus_meta.cc
+++ b/src/kudu/consensus/consensus_meta.cc
@@ -186,6 +186,8 @@ void ConsensusMetadata::MergeCommittedConsensusStatePB(const ConsensusStatePB& c
Status ConsensusMetadata::Flush() {
SCOPED_LOG_SLOW_EXECUTION_PREFIX(WARNING, 500, LogPrefix(), "flushing consensus metadata");
+
+ flush_count_for_tests_++;
// Sanity test to ensure we never write out a bad configuration.
RETURN_NOT_OK_PREPEND(VerifyRaftConfig(pb_.committed_config(), COMMITTED_QUORUM),
"Invalid config in ConsensusMetadata, cannot flush to disk");
@@ -222,7 +224,8 @@ ConsensusMetadata::ConsensusMetadata(FsManager* fs_manager,
: fs_manager_(CHECK_NOTNULL(fs_manager)),
tablet_id_(std::move(tablet_id)),
peer_uuid_(std::move(peer_uuid)),
- has_pending_config_(false) {}
+ has_pending_config_(false),
+ flush_count_for_tests_(0) {}
std::string ConsensusMetadata::LogPrefix() const {
return Substitute("T $0 P $1: ", tablet_id_, peer_uuid_);
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/src/kudu/consensus/consensus_meta.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_meta.h b/src/kudu/consensus/consensus_meta.h
index a8d5f93..9b014fa 100644
--- a/src/kudu/consensus/consensus_meta.h
+++ b/src/kudu/consensus/consensus_meta.h
@@ -138,6 +138,10 @@ class ConsensusMetadata {
// Persist current state of the protobuf to disk.
Status Flush();
+ int flush_count_for_tests() const {
+ return flush_count_for_tests_;
+ }
+
private:
ConsensusMetadata(FsManager* fs_manager, std::string tablet_id,
std::string peer_uuid);
@@ -162,6 +166,9 @@ class ConsensusMetadata {
// Cached role of the peer_uuid_ within the active configuration.
RaftPeerPB::Role active_role_;
+ // The number of times the metadata has been flushed to disk.
+ int flush_count_for_tests_;
+
// Durable fields.
ConsensusMetadataPB pb_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/src/kudu/consensus/raft_consensus.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.cc b/src/kudu/consensus/raft_consensus.cc
index 7a302b5..77125e1 100644
--- a/src/kudu/consensus/raft_consensus.cc
+++ b/src/kudu/consensus/raft_consensus.cc
@@ -378,6 +378,8 @@ Status RaftConsensus::StartElection(ElectionMode mode) {
}
// Increment the term.
+ // TODO: this causes an extra flush of the consensus metadata which
+ // will be flushed again for our vote below. Consolidate these.
RETURN_NOT_OK(IncrementTermUnlocked());
// Snooze to avoid the election timer firing again as much as possible.
@@ -1365,17 +1367,24 @@ Status RaftConsensus::RequestVote(const VoteRequestPB* request, VoteResponsePB*
return RequestVoteRespondAlreadyVotedForOther(request, response);
}
- // The term advanced.
+ // Candidate must have last-logged OpId at least as large as our own to get
+ // our vote.
+ OpId local_last_logged_opid = GetLatestOpIdFromLog();
+ bool vote_yes = !OpIdLessThan(request->candidate_status().last_received(),
+ local_last_logged_opid);
+
+ // Record the term advancement if necessary.
if (request->candidate_term() > state_->GetCurrentTermUnlocked()) {
- RETURN_NOT_OK_PREPEND(HandleTermAdvanceUnlocked(request->candidate_term()),
+ // If we are going to vote for this peer, then we will flush the consensus metadata
+ // to disk below when we record the vote, and we can skip flushing the term advancement
+ // to disk here.
+ auto flush = vote_yes ? ReplicaState::SKIP_FLUSH_TO_DISK : ReplicaState::FLUSH_TO_DISK;
+ RETURN_NOT_OK_PREPEND(HandleTermAdvanceUnlocked(request->candidate_term(), flush),
Substitute("Could not step down in RequestVote. Current term: $0, candidate term: $1",
state_->GetCurrentTermUnlocked(), request->candidate_term()));
}
- // Candidate must have last-logged OpId at least as large as our own to get
- // our vote.
- OpId local_last_logged_opid = GetLatestOpIdFromLog();
- if (OpIdLessThan(request->candidate_status().last_received(), local_last_logged_opid)) {
+ if (!vote_yes) {
return RequestVoteRespondLastOpIdTooOld(local_last_logged_opid, request, response);
}
@@ -2017,7 +2026,8 @@ Status RaftConsensus::IncrementTermUnlocked() {
return HandleTermAdvanceUnlocked(state_->GetCurrentTermUnlocked() + 1);
}
-Status RaftConsensus::HandleTermAdvanceUnlocked(ConsensusTerm new_term) {
+Status RaftConsensus::HandleTermAdvanceUnlocked(ConsensusTerm new_term,
+ ReplicaState::FlushToDisk flush) {
if (new_term <= state_->GetCurrentTermUnlocked()) {
return Status::IllegalState(Substitute("Can't advance term to: $0 current term: $1 is higher.",
new_term, state_->GetCurrentTermUnlocked()));
@@ -2029,7 +2039,7 @@ Status RaftConsensus::HandleTermAdvanceUnlocked(ConsensusTerm new_term) {
}
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Advancing to term " << new_term;
- RETURN_NOT_OK(state_->SetCurrentTermUnlocked(new_term));
+ RETURN_NOT_OK(state_->SetCurrentTermUnlocked(new_term, flush));
term_metric_->set_value(new_term);
return Status::OK();
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/src/kudu/consensus/raft_consensus.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.h b/src/kudu/consensus/raft_consensus.h
index d0f9fe0..90b10b6 100644
--- a/src/kudu/consensus/raft_consensus.h
+++ b/src/kudu/consensus/raft_consensus.h
@@ -29,6 +29,7 @@
#include "kudu/consensus/consensus.pb.h"
#include "kudu/consensus/consensus_meta.h"
#include "kudu/consensus/consensus_queue.h"
+#include "kudu/consensus/raft_consensus_state.h"
#include "kudu/util/atomic.h"
#include "kudu/util/failure_detector.h"
@@ -55,7 +56,6 @@ class ConsensusMetadata;
class Peer;
class PeerProxyFactory;
class PeerManager;
-class ReplicaState;
struct ElectionResult;
class RaftConsensus : public Consensus,
@@ -405,7 +405,10 @@ class RaftConsensus : public Consensus,
Status IncrementTermUnlocked();
// Handle when the term has advanced beyond the current term.
- Status HandleTermAdvanceUnlocked(ConsensusTerm new_term);
+ //
+ // 'flush' may be used to control whether the term change is flushed to disk.
+ Status HandleTermAdvanceUnlocked(ConsensusTerm new_term,
+ ReplicaState::FlushToDisk flush = ReplicaState::FLUSH_TO_DISK);
// Asynchronously (on thread_pool_) notify the tablet peer that the consensus configuration
// has changed, thus reporting it back to the master.
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/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 24428fc..648cb1d 100644
--- a/src/kudu/consensus/raft_consensus_quorum-test.cc
+++ b/src/kudu/consensus/raft_consensus_quorum-test.cc
@@ -1026,6 +1026,10 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
const int kPeerIndex = 1;
scoped_refptr<RaftConsensus> peer;
CHECK_OK(peers_->GetPeerByIdx(kPeerIndex, &peer));
+ auto flush_count = [&]() {
+ return peer->GetReplicaStateForTests()
+ ->consensus_metadata_for_tests()->flush_count_for_tests();
+ };
VoteRequestPB request;
request.set_tablet_id(kTestTablet);
@@ -1033,29 +1037,39 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
// Test that the replica won't vote since it has recently heard from
// a valid leader.
+ int flush_count_before = flush_count();
VoteResponsePB response;
request.set_candidate_uuid("peer-0");
request.set_candidate_term(last_op_id.term() + 1);
ASSERT_OK(peer->RequestVote(&request, &response));
ASSERT_FALSE(response.vote_granted());
ASSERT_EQ(ConsensusErrorPB::LEADER_IS_ALIVE, response.consensus_error().code());
+ ASSERT_EQ(0, flush_count() - flush_count_before)
+ << "A rejected vote should not flush metadata";
// Test that replicas only vote yes for a single peer per term.
// Indicate that replicas should vote even if they think another leader is alive.
// This will allow the rest of the requests in the test to go through.
+ flush_count_before = flush_count();
request.set_ignore_live_leader(true);
ASSERT_OK(peer->RequestVote(&request, &response));
ASSERT_TRUE(response.vote_granted());
ASSERT_EQ(last_op_id.term() + 1, response.responder_term());
ASSERT_NO_FATAL_FAILURE(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 1, "peer-0"));
+ ASSERT_EQ(1, flush_count() - flush_count_before)
+ << "A granted vote should flush only once";
// Ensure we get same response for same term and same UUID.
response.Clear();
+ flush_count_before = flush_count();
ASSERT_OK(peer->RequestVote(&request, &response));
ASSERT_TRUE(response.vote_granted());
+ ASSERT_EQ(0, flush_count() - flush_count_before)
+ << "Confirming a previous vote should not flush";
// Ensure we get a "no" for a different candidate UUID for that term.
+ flush_count_before = flush_count();
response.Clear();
request.set_candidate_uuid("peer-2");
ASSERT_OK(peer->RequestVote(&request, &response));
@@ -1064,6 +1078,8 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
ASSERT_EQ(ConsensusErrorPB::ALREADY_VOTED, response.consensus_error().code());
ASSERT_EQ(last_op_id.term() + 1, response.responder_term());
ASSERT_NO_FATAL_FAILURE(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 1, "peer-0"));
+ ASSERT_EQ(0, flush_count() - flush_count_before)
+ << "Rejected votes for same term should not flush";
//
// Test that replicas refuse votes for an old term.
@@ -1071,6 +1087,7 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
// Increase the term of our candidate, which will cause the voter replica to
// increase its own term to match.
+ flush_count_before = flush_count();
request.set_candidate_uuid("peer-0");
request.set_candidate_term(last_op_id.term() + 2);
response.Clear();
@@ -1078,10 +1095,13 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
ASSERT_TRUE(response.vote_granted());
ASSERT_EQ(last_op_id.term() + 2, response.responder_term());
ASSERT_NO_FATAL_FAILURE(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 2, "peer-0"));
+ ASSERT_EQ(1, flush_count() - flush_count_before)
+ << "Accepted votes with increased term should flush once";
// Now try the old term.
// Note: Use the peer who "won" the election on the previous term (peer-0),
// although in practice the impl does not store historical vote data.
+ flush_count_before = flush_count();
request.set_candidate_term(last_op_id.term() + 1);
response.Clear();
ASSERT_OK(peer->RequestVote(&request, &response));
@@ -1090,11 +1110,14 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
ASSERT_EQ(ConsensusErrorPB::INVALID_TERM, response.consensus_error().code());
ASSERT_EQ(last_op_id.term() + 2, response.responder_term());
ASSERT_NO_FATAL_FAILURE(AssertDurableTermAndVote(kPeerIndex, last_op_id.term() + 2, "peer-0"));
+ ASSERT_EQ(0, flush_count() - flush_count_before)
+ << "Rejected votes for old terms should not flush";
//
// Ensure replicas vote no for an old op index.
//
+ flush_count_before = flush_count();
request.set_candidate_uuid("peer-0");
request.set_candidate_term(last_op_id.term() + 3);
request.mutable_candidate_status()->mutable_last_received()->CopyFrom(MinimumOpId());
@@ -1105,6 +1128,9 @@ TEST_F(RaftConsensusQuorumTest, TestRequestVote) {
ASSERT_EQ(ConsensusErrorPB::LAST_OPID_TOO_OLD, response.consensus_error().code());
ASSERT_EQ(last_op_id.term() + 3, response.responder_term());
ASSERT_NO_FATAL_FAILURE(AssertDurableTermWithoutVote(kPeerIndex, last_op_id.term() + 3));
+ ASSERT_EQ(1, flush_count() - flush_count_before)
+ << "Rejected votes for old op index but new term should flush once.";
+
// Send a "heartbeat" to the peer. It should be rejected.
ConsensusRequestPB req;
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/src/kudu/consensus/raft_consensus_state.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus_state.cc b/src/kudu/consensus/raft_consensus_state.cc
index 92222bf..7763cb6 100644
--- a/src/kudu/consensus/raft_consensus_state.cc
+++ b/src/kudu/consensus/raft_consensus_state.cc
@@ -280,7 +280,8 @@ bool ReplicaState::IsOpCommittedOrPending(const OpId& op_id, bool* term_mismatch
return true;
}
-Status ReplicaState::SetCurrentTermUnlocked(int64_t new_term) {
+Status ReplicaState::SetCurrentTermUnlocked(int64_t new_term,
+ FlushToDisk flush) {
TRACE_EVENT1("consensus", "ReplicaState::SetCurrentTermUnlocked",
"term", new_term);
DCHECK(update_lock_.is_locked());
@@ -291,7 +292,9 @@ Status ReplicaState::SetCurrentTermUnlocked(int64_t new_term) {
}
cmeta_->set_current_term(new_term);
cmeta_->clear_voted_for();
- CHECK_OK(cmeta_->Flush());
+ if (flush == FLUSH_TO_DISK) {
+ CHECK_OK(cmeta_->Flush());
+ }
ClearLeaderUnlocked();
last_received_op_id_current_leader_ = MinimumOpId();
return Status::OK();
http://git-wip-us.apache.org/repos/asf/kudu/blob/d87e3ff6/src/kudu/consensus/raft_consensus_state.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus_state.h b/src/kudu/consensus/raft_consensus_state.h
index efcaff2..f73cccd 100644
--- a/src/kudu/consensus/raft_consensus_state.h
+++ b/src/kudu/consensus/raft_consensus_state.h
@@ -194,9 +194,19 @@ class ReplicaState {
// otherwise return the committed configuration.
const RaftConfigPB& GetActiveConfigUnlocked() const;
+ // Enum for the 'flush' argument to SetCurrentTermUnlocked() below.
+ enum FlushToDisk {
+ SKIP_FLUSH_TO_DISK,
+ FLUSH_TO_DISK
+ };
+
// Checks if the term change is legal. If so, sets 'current_term'
// to 'new_term' and sets 'has voted' to no for the current term.
- Status SetCurrentTermUnlocked(int64_t new_term) WARN_UNUSED_RESULT;
+ //
+ // If the caller knows that it will call another method soon after
+ // to flush the change to disk, it may set 'flush' to 'SKIP_FLUSH_TO_DISK'.
+ Status SetCurrentTermUnlocked(int64_t new_term,
+ FlushToDisk flush) WARN_UNUSED_RESULT;
// Returns the term set in the last config change round.
const int64_t GetCurrentTermUnlocked() const;
@@ -326,6 +336,10 @@ class ReplicaState {
// The update_lock_ must be held.
ReplicaState::State state() const;
+ ConsensusMetadata* consensus_metadata_for_tests() {
+ return cmeta_.get();
+ }
+
private:
const ConsensusOptions options_;
[2/3] kudu git commit: KUDU-1623. Properly handle UPSERTS that only
include PK column
Posted by to...@apache.org.
KUDU-1623. Properly handle UPSERTS that only include PK column
This fixes a bug reported by Chris George in which the tablet server
would crash when handling an UPSERT which was converted into an empty
UPDATE. For example, in pseudo-SQL:
CREATE TABLE t (
x INT NOT NULL PRIMARY KEY
);
UPSERT INTO t VALUES (1);
UPSERT INTO t VALUES (1);
Here the second upsert would detect the primary key already existed and
convert into an update containing any non-PK columns. Since there are no
non-PK columns, the update was "empty", and ended up as an empty
RowChangeList. In DEBUG builds, this fired a DCHECK, but in RELEASE
builds, we ended up with an empty RowChangeList in the DMS. At flush
time, this would cause other assertions to fire or crash.
The fix is relatively simple: if we see that an upsert converted into an
empty update, we just drop the delta rather than continuing to insert
into the delta store. This required loosing a check at bootstrap time,
where we previously assumed that every mutation must have been recorded
in at least one store.
The test changes are much larger than the fix itself due to some
refactoring. The main gist of the change is to introduce operations in
tablet_random_access-test and fuzz-itest where we send an UPSERT which
contains only the primary key column. These tests crashed prior to the
fix and pass now.
Change-Id: Ic878f6e51ead5bf91335ddb47536e7c29de11ac4
Reviewed-on: http://gerrit.cloudera.org:8080/4441
Reviewed-by: David Ribeiro Alves <dr...@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/9911c489
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/9911c489
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/9911c489
Branch: refs/heads/master
Commit: 9911c489c45b3a261ee50ad1f837387b4953421b
Parents: 4ba39e4
Author: Todd Lipcon <to...@apache.org>
Authored: Fri Sep 16 11:14:07 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Mon Sep 19 21:10:54 2016 +0000
----------------------------------------------------------------------
src/kudu/integration-tests/fuzz-itest.cc | 93 ++++++++++-----
src/kudu/tablet/key_value_test_schema.h | 60 ++++++++++
src/kudu/tablet/tablet.cc | 11 +-
src/kudu/tablet/tablet_bootstrap.cc | 7 +-
src/kudu/tablet/tablet_random_access-test.cc | 131 +++++++++++++++-------
5 files changed, 229 insertions(+), 73 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/9911c489/src/kudu/integration-tests/fuzz-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/fuzz-itest.cc b/src/kudu/integration-tests/fuzz-itest.cc
index 5629e14..2134a56 100644
--- a/src/kudu/integration-tests/fuzz-itest.cc
+++ b/src/kudu/integration-tests/fuzz-itest.cc
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+#include <boost/optional.hpp>
+#include <boost/optional/optional_io.hpp>
#include <glog/logging.h>
#include <glog/stl_logging.h>
#include <gtest/gtest.h>
@@ -31,6 +33,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/master/mini_master.h"
#include "kudu/integration-tests/mini_cluster.h"
+#include "kudu/tablet/key_value_test_schema.h"
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet_peer.h"
#include "kudu/tserver/mini_tablet_server.h"
@@ -43,6 +46,7 @@ DEFINE_int32(keyspace_size, 2, "number of distinct primary keys to test with");
DECLARE_bool(enable_maintenance_manager);
DECLARE_bool(use_hybrid_clock);
+using boost::optional;
using std::string;
using std::vector;
using std::unique_ptr;
@@ -52,6 +56,7 @@ using std::unique_ptr;
enum TestOpType {
TEST_INSERT,
TEST_UPSERT,
+ TEST_UPSERT_PK_ONLY,
TEST_UPDATE,
TEST_DELETE,
TEST_FLUSH_OPS,
@@ -93,6 +98,7 @@ namespace tablet {
const char* TestOpType_names[] = {
"TEST_INSERT",
"TEST_UPSERT",
+ "TEST_UPSERT_PK_ONLY",
"TEST_UPDATE",
"TEST_DELETE",
"TEST_FLUSH_OPS",
@@ -129,10 +135,7 @@ class FuzzTest : public KuduTest {
FLAGS_enable_maintenance_manager = false;
FLAGS_use_hybrid_clock = false;
- KuduSchemaBuilder b;
- b.AddColumn("key")->Type(KuduColumnSchema::INT32)->NotNull()->PrimaryKey();
- b.AddColumn("val")->Type(KuduColumnSchema::INT32);
- CHECK_OK(b.Build(&schema_));
+ schema_ = client::KuduSchemaFromSchema(CreateKeyValueTestSchema());
}
void SetUp() override {
@@ -192,9 +195,12 @@ class FuzzTest : public KuduTest {
return tablet_peer_->tablet();
}
- // Adds an insert for the given key/value pair to 'ops', returning the new stringified
- // value of the row.
- string InsertOrUpsertRow(int key, int val, TestOpType type) {
+ // Adds an insert for the given key/value pair to 'ops', returning the new contents
+ // of the row.
+ ExpectedKeyValueRow InsertOrUpsertRow(int key, int val,
+ optional<ExpectedKeyValueRow> old_row,
+ TestOpType type) {
+ ExpectedKeyValueRow ret;
unique_ptr<KuduWriteOperation> op;
if (type == TEST_INSERT) {
op.reset(table_->NewInsert());
@@ -203,45 +209,54 @@ class FuzzTest : public KuduTest {
}
KuduPartialRow* row = op->mutable_row();
CHECK_OK(row->SetInt32(0, key));
- if (val & 1) {
- CHECK_OK(row->SetNull(1));
+ ret.key = key;
+ if (type != TEST_UPSERT_PK_ONLY) {
+ if (val & 1) {
+ CHECK_OK(row->SetNull(1));
+ } else {
+ CHECK_OK(row->SetInt32(1, val));
+ ret.val = val;
+ }
} else {
- CHECK_OK(row->SetInt32(1, val));
+ // For "upsert PK only", we expect the row to keep its old value
+ // the row existed, or NULL if there was no old row.
+ ret.val = old_row ? old_row->val : boost::none;
}
- string ret = row->ToString();
CHECK_OK(session_->Apply(op.release()));
return ret;
}
- // Adds an update of the given key/value pair to 'ops', returning the new stringified
- // value of the row.
- string MutateRow(int key, uint32_t new_val) {
+ // Adds an update of the given key/value pair to 'ops', returning the new contents
+ // of the row.
+ ExpectedKeyValueRow MutateRow(int key, uint32_t new_val) {
+ ExpectedKeyValueRow ret;
unique_ptr<KuduUpdate> update(table_->NewUpdate());
KuduPartialRow* row = update->mutable_row();
CHECK_OK(row->SetInt32(0, key));
+ ret.key = key;
if (new_val & 1) {
CHECK_OK(row->SetNull(1));
} else {
CHECK_OK(row->SetInt32(1, new_val));
+ ret.val = new_val;
}
- string ret = row->ToString();
CHECK_OK(session_->Apply(update.release()));
return ret;
}
- // Adds a delete of the given row to 'ops', returning an empty string (indicating that
+ // Adds a delete of the given row to 'ops', returning boost::none (indicating that
// the row no longer exists).
- string DeleteRow(int key) {
+ optional<ExpectedKeyValueRow> DeleteRow(int key) {
unique_ptr<KuduDelete> del(table_->NewDelete());
KuduPartialRow* row = del->mutable_row();
CHECK_OK(row->SetInt32(0, key));
CHECK_OK(session_->Apply(del.release()));
- return "";
+ return boost::none;
}
// Random-read the given row, returning its current value.
- // If the row doesn't exist, returns "()".
- string GetRow(int key) {
+ // If the row doesn't exist, returns boost::none.
+ optional<ExpectedKeyValueRow> GetRow(int key) {
KuduScanner s(table_.get());
CHECK_OK(s.AddConjunctPredicate(table_->NewComparisonPredicate(
"key", KuduPredicate::EQUAL, KuduValue::FromInt(key))));
@@ -250,10 +265,16 @@ class FuzzTest : public KuduTest {
KuduScanBatch batch;
CHECK_OK(s.NextBatch(&batch));
for (KuduScanBatch::RowPtr row : batch) {
- return row.ToString();
+ ExpectedKeyValueRow ret;
+ CHECK_OK(row.GetInt32(0, &ret.key));
+ if (!row.IsNull(1)) {
+ ret.val = 0;
+ CHECK_OK(row.GetInt32(1, ret.val.get_ptr()));
+ }
+ return ret;
}
}
- return "()";
+ return boost::none;
}
protected:
@@ -290,7 +311,8 @@ void GenerateTestCase(vector<TestOp>* ops, int len) {
data_in_mrs = true;
break;
case TEST_UPSERT:
- ops->push_back({TEST_UPSERT, row_key});
+ case TEST_UPSERT_PK_ONLY:
+ ops->push_back({r, row_key});
exists[row_key] = true;
ops_pending = true;
// If the row doesn't currently exist, this will act like an insert
@@ -387,20 +409,23 @@ void FuzzTest::RunFuzzCase(const vector<TestOp>& test_ops,
// into a test method in order to reproduce a failure.
LOG(INFO) << "test case:\n" << DumpTestCase(test_ops);
- vector<string> cur_val(FLAGS_keyspace_size);
- vector<string> pending_val(FLAGS_keyspace_size);
+ vector<optional<ExpectedKeyValueRow>> cur_val(FLAGS_keyspace_size);
+ vector<optional<ExpectedKeyValueRow>> pending_val(FLAGS_keyspace_size);
int i = 0;
for (const TestOp& test_op : test_ops) {
- string val_in_table = GetRow(test_op.row_key);
- EXPECT_EQ("(" + cur_val[test_op.row_key] + ")", val_in_table);
+ optional<ExpectedKeyValueRow> val_in_table = GetRow(test_op.row_key);
+ EXPECT_EQ(cur_val[test_op.row_key], val_in_table);
LOG(INFO) << test_op.ToString();
switch (test_op.type) {
case TEST_INSERT:
case TEST_UPSERT:
- pending_val[test_op.row_key] = InsertOrUpsertRow(test_op.row_key, i++, test_op.type);
+ case TEST_UPSERT_PK_ONLY: {
+ pending_val[test_op.row_key] = InsertOrUpsertRow(
+ test_op.row_key, i++, pending_val[test_op.row_key], test_op.type);
break;
+ }
case TEST_UPDATE:
for (int j = 0; j < update_multiplier; j++) {
pending_val[test_op.row_key] = MutateRow(test_op.row_key, i++);
@@ -669,5 +694,17 @@ TEST_F(FuzzTest, TestUpsertSeq) {
});
}
+// Regression test for KUDU-1623: updates without primary key
+// columns specified can cause crashes and issues at restart.
+TEST_F(FuzzTest, TestUpsert_PKOnly) {
+ RunFuzzCase({
+ {TEST_INSERT, 1},
+ {TEST_FLUSH_OPS, 0},
+ {TEST_UPSERT_PK_ONLY, 1},
+ {TEST_FLUSH_OPS, 0},
+ {TEST_RESTART_TS, 0}
+ });
+}
+
} // namespace tablet
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/9911c489/src/kudu/tablet/key_value_test_schema.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/key_value_test_schema.h b/src/kudu/tablet/key_value_test_schema.h
new file mode 100644
index 0000000..e81a4f2
--- /dev/null
+++ b/src/kudu/tablet/key_value_test_schema.h
@@ -0,0 +1,60 @@
+// 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.
+
+// This file contains a schema and some utility code for a simple
+// "Key-Value" table, with int32 keys and nullable int32 values. This
+// is used by a few tests.
+#pragma once
+
+#include <boost/optional.hpp>
+#include <boost/optional/optional_io.hpp>
+#include <iostream>
+
+#include "kudu/common/schema.h"
+
+namespace kudu {
+
+// Struct for keeping track of what we think the contents of a row should be after
+// our random mutations.
+struct ExpectedKeyValueRow {
+ // Non-nullable key.
+ int32_t key;
+
+ // Nullable value.
+ boost::optional<int32_t> val;
+
+ bool operator==(const ExpectedKeyValueRow& other) const {
+ return key == other.key && val == other.val;
+ }
+};
+
+inline Schema CreateKeyValueTestSchema() {
+ return Schema({ColumnSchema("key", INT32),
+ ColumnSchema("val", INT32, true) }, 1);
+}
+
+inline std::ostream& operator<<(std::ostream& o, const ExpectedKeyValueRow& t) {
+ o << "{" << t.key << ", ";
+ if (t.val == boost::none) {
+ o << "NULL";
+ } else {
+ o << *t.val;
+ }
+ return o << "}";
+}
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/9911c489/src/kudu/tablet/tablet.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index 508aef7..8b7e0e4 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -457,9 +457,18 @@ Status Tablet::ApplyUpsertAsUpdate(WriteTransactionState* tx_state,
enc.AddColumnUpdate(c, schema->column_id(i), val);
}
+ // If the UPSERT just included the primary key columns, and the rest
+ // were unset (eg because the table only _has_ primary keys, or because
+ // the rest are intended to be set to their defaults), we need to
+ // avoid doing anything.
+ gscoped_ptr<OperationResultPB> result(new OperationResultPB());
+ if (enc.is_empty()) {
+ upsert->SetMutateSucceeded(std::move(result));
+ return Status::OK();
+ }
+
RowChangeList rcl = enc.as_changelist();
- gscoped_ptr<OperationResultPB> result(new OperationResultPB());
Status s = rowset->MutateRow(tx_state->timestamp(),
*upsert->key_probe,
rcl,
http://git-wip-us.apache.org/repos/asf/kudu/blob/9911c489/src/kudu/tablet/tablet_bootstrap.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_bootstrap.cc b/src/kudu/tablet/tablet_bootstrap.cc
index 845f4af..8650e81 100644
--- a/src/kudu/tablet/tablet_bootstrap.cc
+++ b/src/kudu/tablet/tablet_bootstrap.cc
@@ -1448,10 +1448,13 @@ Status TabletBootstrap::FilterOperation(const OperationResultPB& op_result,
}
int num_mutated_stores = op_result.mutated_stores_size();
- if (PREDICT_FALSE(num_mutated_stores == 0 || num_mutated_stores > 2)) {
- return Status::Corruption(Substitute("All operations must have one or two mutated_stores: $0",
+ if (PREDICT_FALSE(num_mutated_stores > 2)) {
+ return Status::Corruption(Substitute("All operations must have at most two mutated_stores: $0",
op_result.ShortDebugString()));
}
+ // NOTE: it's possible that num_mutated_stores = 0 in the case of an
+ // UPSERT which only specified the primary key. In that case, if the
+ // row already existed, it gets dropped without converting into an UPDATE.
// The mutation may have been duplicated, so we'll check whether any of the
// output targets was active.
http://git-wip-us.apache.org/repos/asf/kudu/blob/9911c489/src/kudu/tablet/tablet_random_access-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_random_access-test.cc b/src/kudu/tablet/tablet_random_access-test.cc
index ae9822c..8d9304f 100644
--- a/src/kudu/tablet/tablet_random_access-test.cc
+++ b/src/kudu/tablet/tablet_random_access-test.cc
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+#include <boost/optional.hpp>
+#include <boost/optional/optional_io.hpp>
#include <glog/logging.h>
#include <glog/stl_logging.h>
#include <gtest/gtest.h>
@@ -24,6 +26,7 @@
#include "kudu/common/schema.h"
#include "kudu/gutil/casts.h"
+#include "kudu/tablet/key_value_test_schema.h"
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet-test-base.h"
#include "kudu/util/stopwatch.h"
@@ -36,6 +39,7 @@ DEFINE_int32(update_delete_ratio, 4, "ratio of update:delete when mutating exist
DECLARE_int32(deltafile_default_block_size);
+using boost::optional;
using std::string;
using std::vector;
@@ -51,9 +55,8 @@ namespace tablet {
class TestRandomAccess : public KuduTabletTest {
public:
TestRandomAccess()
- : KuduTabletTest(Schema({ ColumnSchema("key", INT32),
- ColumnSchema("val", INT32, true) }, 1)),
- done_(1) {
+ : KuduTabletTest(CreateKeyValueTestSchema()),
+ done_(1) {
OverrideFlagForSlowTests("keyspace_size", "30000");
OverrideFlagForSlowTests("runtime_seconds", "10");
OverrideFlagForSlowTests("sleep_between_background_ops_ms", "1000");
@@ -67,6 +70,7 @@ class TestRandomAccess : public KuduTabletTest {
virtual void SetUp() OVERRIDE {
KuduTabletTest::SetUp();
writer_.reset(new LocalTabletWriter(tablet().get(), &client_schema_));
+ SeedRandom();
}
// Pick a random row of the table, verify its current state, and then
@@ -81,21 +85,21 @@ class TestRandomAccess : public KuduTabletTest {
// and validates the correct errors.
void DoRandomBatch() {
int key = rand() % expected_tablet_state_.size();
- string& cur_val = expected_tablet_state_[key];
+ optional<ExpectedKeyValueRow>& cur_val = expected_tablet_state_[key];
// Check that a read yields what we expect.
- string val_in_table = GetRow(key);
- ASSERT_EQ("(" + cur_val + ")", val_in_table);
+ optional<ExpectedKeyValueRow> val_in_table = GetRow(key);
+ ASSERT_EQ(cur_val, val_in_table);
vector<LocalTabletWriter::Op> pending;
for (int i = 0; i < 3; i++) {
int new_val = rand();
- if (cur_val.empty()) {
+ if (cur_val == boost::none) {
// If there is no row, then randomly insert or upsert.
if (rand() % 2 == 1) {
cur_val = InsertRow(key, new_val, &pending);
} else {
- cur_val = UpsertRow(key, new_val, &pending);
+ cur_val = UpsertRow(key, new_val, cur_val, &pending);
}
} else {
if (new_val % (FLAGS_update_delete_ratio + 1) == 0) {
@@ -104,13 +108,19 @@ class TestRandomAccess : public KuduTabletTest {
// If we are meant to update an existing row, randomly choose
// between update and upsert.
if (rand() % 2 == 1) {
- cur_val = MutateRow(key, new_val, &pending);
+ cur_val = MutateRow(key, new_val, cur_val, &pending);
} else {
- cur_val = UpsertRow(key, new_val, &pending);
+ cur_val = UpsertRow(key, new_val, cur_val, &pending);
}
}
}
}
+
+ VLOG(1) << "Performing batch:";
+ for (const auto& op : pending) {
+ VLOG(1) << RowOperationsPB::Type_Name(op.type) << " " << op.row->ToString();
+ }
+
CHECK_OK(writer_->WriteBatch(pending));
for (LocalTabletWriter::Op op : pending) {
delete op.row;
@@ -153,55 +163,88 @@ class TestRandomAccess : public KuduTabletTest {
// Adds an insert for the given key/value pair to 'ops', returning the new stringified
// value of the row.
- string InsertRow(int key, int val, vector<LocalTabletWriter::Op>* ops) {
- return InsertOrUpsertRow(RowOperationsPB::INSERT, key, val, ops);
+ optional<ExpectedKeyValueRow> InsertRow(int key, int val, vector<LocalTabletWriter::Op>* ops) {
+ return DoRowOp(RowOperationsPB::INSERT, key, val, boost::none, ops);
}
- string UpsertRow(int key, int val, vector<LocalTabletWriter::Op>* ops) {
- return InsertOrUpsertRow(RowOperationsPB::UPSERT, key, val, ops);
- }
-
- string InsertOrUpsertRow(RowOperationsPB::Type type, int key, int val,
- vector<LocalTabletWriter::Op>* ops) {
- gscoped_ptr<KuduPartialRow> row(new KuduPartialRow(&client_schema_));
- CHECK_OK(row->SetInt32(0, key));
- if (val & 1) {
- CHECK_OK(row->SetNull(1));
- } else {
- CHECK_OK(row->SetInt32(1, val));
- }
- string ret = row->ToString();
- ops->push_back(LocalTabletWriter::Op(type, row.release()));
- return ret;
+ optional<ExpectedKeyValueRow> UpsertRow(int key,
+ int val,
+ const optional<ExpectedKeyValueRow>& old_row,
+ vector<LocalTabletWriter::Op>* ops) {
+ return DoRowOp(RowOperationsPB::UPSERT, key, val, old_row, ops);
}
// Adds an update of the given key/value pair to 'ops', returning the new stringified
// value of the row.
- string MutateRow(int key, uint32_t new_val, vector<LocalTabletWriter::Op>* ops) {
+ optional<ExpectedKeyValueRow> MutateRow(int key,
+ uint32_t new_val,
+ const optional<ExpectedKeyValueRow>& old_row,
+ vector<LocalTabletWriter::Op>* ops) {
+ return DoRowOp(RowOperationsPB::UPDATE, key, new_val, old_row, ops);
+ }
+
+ optional<ExpectedKeyValueRow> DoRowOp(RowOperationsPB::Type type,
+ int key,
+ int val,
+ const optional<ExpectedKeyValueRow>& old_row,
+ vector<LocalTabletWriter::Op>* ops) {
+
gscoped_ptr<KuduPartialRow> row(new KuduPartialRow(&client_schema_));
CHECK_OK(row->SetInt32(0, key));
- if (new_val & 1) {
- CHECK_OK(row->SetNull(1));
- } else {
- CHECK_OK(row->SetInt32(1, new_val));
+ optional<ExpectedKeyValueRow> ret = ExpectedKeyValueRow();
+ ret->key = key;
+
+ switch (type) {
+ case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPDATE:
+ case RowOperationsPB::INSERT:
+ switch (val % 2) {
+ case 0:
+ CHECK_OK(row->SetNull(1));
+ ret->val = boost::none;
+ break;
+ case 1:
+ CHECK_OK(row->SetInt32(1, val));
+ ret->val = val;
+ break;
+ }
+
+ if ((type != RowOperationsPB::UPDATE) && (val % 3 == 1)) {
+ // Don't set the value. In the case of an INSERT or an UPSERT with no pre-existing
+ // row, this should default to NULL. Otherwise it should remain set to whatever it
+ // was previously set to.
+ CHECK_OK(row->Unset(1));
+
+ if (type == RowOperationsPB::INSERT || old_row == boost::none) {
+ ret->val = boost::none;
+ } else {
+ ret->val = old_row->val;
+ }
+ }
+ break;
+ case RowOperationsPB::DELETE:
+ ret = boost::none;
+ break;
+ default:
+ LOG(FATAL) << "Unknown type: " << type;
}
- string ret = row->ToString();
- ops->push_back(LocalTabletWriter::Op(RowOperationsPB::UPDATE, row.release()));
+ ops->push_back(LocalTabletWriter::Op(type, row.release()));
return ret;
}
+
// Adds a delete of the given row to 'ops', returning an empty string (indicating that
// the row no longer exists).
- string DeleteRow(int key, vector<LocalTabletWriter::Op>* ops) {
+ optional<ExpectedKeyValueRow> DeleteRow(int key, vector<LocalTabletWriter::Op>* ops) {
gscoped_ptr<KuduPartialRow> row(new KuduPartialRow(&client_schema_));
CHECK_OK(row->SetInt32(0, key));
ops->push_back(LocalTabletWriter::Op(RowOperationsPB::DELETE, row.release()));
- return "";
+ return boost::none;
}
// Random-read the given row, returning its current value.
- // If the row doesn't exist, returns "()".
- string GetRow(int key) {
+ // If the row doesn't exist, returns boost::none.
+ optional<ExpectedKeyValueRow> GetRow(int key) {
ScanSpec spec;
const Schema& schema = this->client_schema_;
gscoped_ptr<RowwiseIterator> iter;
@@ -210,7 +253,7 @@ class TestRandomAccess : public KuduTabletTest {
spec.AddPredicate(pred_one);
CHECK_OK(iter->Init(&spec));
- string ret = "()";
+ optional<ExpectedKeyValueRow> ret;
int n_results = 0;
Arena arena(1024, 4*1024*1024);
@@ -229,7 +272,11 @@ class TestRandomAccess : public KuduTabletTest {
<< " and now have new matching row: "
<< schema.DebugRow(block.row(i))
<< " iterator: " << iter->ToString();
- ret = schema.DebugRow(block.row(i));
+ ret = ExpectedKeyValueRow();
+ ret->key = *schema.ExtractColumnFromRow<INT32>(block.row(i), 0);
+ if (!block.row(i).is_null(1)) {
+ ret->val = *schema.ExtractColumnFromRow<INT32>(block.row(i), 1);
+ }
n_results++;
}
}
@@ -238,7 +285,7 @@ class TestRandomAccess : public KuduTabletTest {
protected:
// The current expected state of the tablet.
- vector<string> expected_tablet_state_;
+ vector<optional<ExpectedKeyValueRow>> expected_tablet_state_;
// Latch triggered when the main thread is finished performing
// operations. This stops the compact/flush thread.