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/05/19 06:56:32 UTC
[1/2] incubator-kudu git commit: KUDU-1002. Add support for UPSERT
Repository: incubator-kudu
Updated Branches:
refs/heads/master efe40a334 -> 591c9ccb2
KUDU-1002. Add support for UPSERT
This adds support for an Upsert operation to the server side and
to the C++ client. A follow-on patch will add support on the Java
client.
Change-Id: Ic83d40bb7d577509d64aa7986f9dcd0280400c09
Reviewed-on: http://gerrit.cloudera.org:8080/3101
Reviewed-by: Dan Burkert <da...@cloudera.com>
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/56c43158
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/56c43158
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/56c43158
Branch: refs/heads/master
Commit: 56c431585ed7ad07ef53d4d8ad254f855769749a
Parents: efe40a3
Author: Todd Lipcon <to...@apache.org>
Authored: Tue May 17 00:26:48 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Thu May 19 01:55:31 2016 +0000
----------------------------------------------------------------------
src/kudu/client/client-test.cc | 81 +++++++++++++++++++
src/kudu/client/client.cc | 4 +
src/kudu/client/client.h | 7 ++
src/kudu/client/write_op.cc | 10 +++
src/kudu/client/write_op.h | 55 +++++++++----
src/kudu/common/row_operations-test.cc | 9 ++-
src/kudu/common/row_operations.cc | 22 ++++--
src/kudu/common/row_operations.h | 13 +++-
src/kudu/common/wire_protocol.proto | 4 +-
src/kudu/integration-tests/fuzz-itest.cc | 36 +++++++--
src/kudu/tablet/local_tablet_writer.h | 4 +
src/kudu/tablet/tablet-test-base.h | 23 +++++-
src/kudu/tablet/tablet-test.cc | 22 ++++++
src/kudu/tablet/tablet.cc | 82 ++++++++++++++++----
src/kudu/tablet/tablet.h | 28 ++++---
src/kudu/tablet/tablet_bootstrap.cc | 4 +-
src/kudu/tablet/tablet_metrics.cc | 4 +
src/kudu/tablet/tablet_metrics.h | 1 +
src/kudu/tablet/tablet_random_access-test.cc | 27 ++++++-
src/kudu/tablet/transactions/transaction.cc | 1 +
src/kudu/tablet/transactions/transaction.h | 1 +
.../tablet/transactions/write_transaction.cc | 4 +
22 files changed, 372 insertions(+), 70 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/client/client-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/client-test.cc b/src/kudu/client/client-test.cc
index e3ce933..8d455e5 100644
--- a/src/kudu/client/client-test.cc
+++ b/src/kudu/client/client-test.cc
@@ -1475,6 +1475,18 @@ static Status ApplyInsertToSession(KuduSession* session,
return session->Apply(insert.release());
}
+static Status ApplyUpsertToSession(KuduSession* session,
+ const shared_ptr<KuduTable>& table,
+ int row_key,
+ int int_val,
+ const char* string_val) {
+ gscoped_ptr<KuduUpsert> upsert(table->NewUpsert());
+ RETURN_NOT_OK(upsert->mutable_row()->SetInt32("key", row_key));
+ RETURN_NOT_OK(upsert->mutable_row()->SetInt32("int_val", int_val));
+ RETURN_NOT_OK(upsert->mutable_row()->SetStringCopy("string_val", string_val));
+ return session->Apply(upsert.release());
+}
+
static Status ApplyUpdateToSession(KuduSession* session,
const shared_ptr<KuduTable>& table,
int row_key,
@@ -1877,6 +1889,75 @@ TEST_F(ClientTest, TestMutateNonexistentRow) {
ASSERT_EQ(0, rows.size());
}
+TEST_F(ClientTest, TestUpsert) {
+ shared_ptr<KuduSession> session = client_->NewSession();
+ ASSERT_OK(session->SetFlushMode(KuduSession::MANUAL_FLUSH));
+
+ // Perform and verify UPSERT which acts as an INSERT.
+ ASSERT_OK(ApplyUpsertToSession(session.get(), client_table_, 1, 1, "original row"));
+ FlushSessionOrDie(session);
+
+ {
+ vector<string> rows;
+ ScanTableToStrings(client_table_.get(), &rows);
+ EXPECT_EQ(vector<string>({"(int32 key=1, int32 int_val=1, string string_val=original row, "
+ "int32 non_null_with_default=12345)"}),
+ rows);
+ }
+
+ // Perform and verify UPSERT which acts as an UPDATE.
+ ASSERT_OK(ApplyUpsertToSession(session.get(), client_table_, 1, 2, "upserted row"));
+ FlushSessionOrDie(session);
+
+ {
+ vector<string> rows;
+ ScanTableToStrings(client_table_.get(), &rows);
+ EXPECT_EQ(vector<string>({"(int32 key=1, int32 int_val=2, string string_val=upserted row, "
+ "int32 non_null_with_default=12345)"}),
+ rows);
+ }
+
+ // Apply an UPDATE including the column that has a default and verify it.
+ {
+ gscoped_ptr<KuduUpdate> update(client_table_->NewUpdate());
+ KuduPartialRow* row = update->mutable_row();
+ ASSERT_OK(row->SetInt32("key", 1));
+ ASSERT_OK(row->SetStringCopy("string_val", "updated row"));
+ ASSERT_OK(row->SetInt32("non_null_with_default", 999));
+ ASSERT_OK(session->Apply(update.release()));
+ FlushSessionOrDie(session);
+ }
+ {
+ vector<string> rows;
+ ScanTableToStrings(client_table_.get(), &rows);
+ EXPECT_EQ(vector<string>({"(int32 key=1, int32 int_val=2, string string_val=updated row, "
+ "int32 non_null_with_default=999)"}),
+ rows);
+ }
+
+ // Perform another UPSERT. This upsert doesn't specify the 'non_null_with_default'
+ // column, and therefore should not revert it back to its default.
+ ASSERT_OK(ApplyUpsertToSession(session.get(), client_table_, 1, 3, "upserted row 2"));
+ FlushSessionOrDie(session);
+ {
+ vector<string> rows;
+ ScanTableToStrings(client_table_.get(), &rows);
+ EXPECT_EQ(vector<string>({"(int32 key=1, int32 int_val=3, string string_val=upserted row 2, "
+ "int32 non_null_with_default=999)"}),
+ rows);
+ }
+
+ // Delete the row.
+ ASSERT_OK(ApplyDeleteToSession(session.get(), client_table_, 1));
+ FlushSessionOrDie(session);
+ {
+ vector<string> rows;
+ ScanTableToStrings(client_table_.get(), &rows);
+ EXPECT_EQ(vector<string>({}), rows);
+ }
+}
+
+
TEST_F(ClientTest, TestWriteWithBadColumn) {
shared_ptr<KuduTable> table;
ASSERT_OK(client_->OpenTable(kTableName, &table));
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/client/client.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/client.cc b/src/kudu/client/client.cc
index 71ae78a..7a16004 100644
--- a/src/kudu/client/client.cc
+++ b/src/kudu/client/client.cc
@@ -563,6 +563,10 @@ KuduInsert* KuduTable::NewInsert() {
return new KuduInsert(shared_from_this());
}
+KuduUpsert* KuduTable::NewUpsert() {
+ return new KuduUpsert(shared_from_this());
+}
+
KuduUpdate* KuduTable::NewUpdate() {
return new KuduUpdate(shared_from_this());
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/client/client.h
----------------------------------------------------------------------
diff --git a/src/kudu/client/client.h b/src/kudu/client/client.h
index 6ff0d65..371d3cb 100644
--- a/src/kudu/client/client.h
+++ b/src/kudu/client/client.h
@@ -417,7 +417,14 @@ class KUDU_EXPORT KuduTable : public sp::enable_shared_from_this<KuduTable> {
// Create a new write operation for this table. It is the caller's
// responsibility to free it, unless it is passed to KuduSession::Apply().
+ //
+ // The provided operations are:
+ // INSERT: Adds a new row. Fails if the row already exists.
+ // UPSERT: Adds a new row. If there's an existing row, updates it.
+ // UPDATE: Updates an existing row. Fails if the row does not exist.
+ // DELETE: Deletes an existing row. Fails if the row does not exist.
KuduInsert* NewInsert();
+ KuduUpsert* NewUpsert();
KuduUpdate* NewUpdate();
KuduDelete* NewDelete();
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/client/write_op.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/write_op.cc b/src/kudu/client/write_op.cc
index 6064885..25891d9 100644
--- a/src/kudu/client/write_op.cc
+++ b/src/kudu/client/write_op.cc
@@ -32,6 +32,7 @@ RowOperationsPB_Type ToInternalWriteType(KuduWriteOperation::Type type) {
case KuduWriteOperation::INSERT: return RowOperationsPB_Type_INSERT;
case KuduWriteOperation::UPDATE: return RowOperationsPB_Type_UPDATE;
case KuduWriteOperation::DELETE: return RowOperationsPB_Type_DELETE;
+ case KuduWriteOperation::UPSERT: return RowOperationsPB_Type_UPSERT;
default: LOG(FATAL) << "Unexpected write operation type: " << type;
}
}
@@ -104,5 +105,14 @@ KuduDelete::KuduDelete(const shared_ptr<KuduTable>& table)
KuduDelete::~KuduDelete() {}
+// Upsert -----------------------------------------------------------------------
+
+KuduUpsert::KuduUpsert(const shared_ptr<KuduTable>& table)
+ : KuduWriteOperation(table) {
+}
+
+KuduUpsert::~KuduUpsert() {}
+
+
} // namespace client
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/client/write_op.h
----------------------------------------------------------------------
diff --git a/src/kudu/client/write_op.h b/src/kudu/client/write_op.h
index 4340eb3..51e16a0 100644
--- a/src/kudu/client/write_op.h
+++ b/src/kudu/client/write_op.h
@@ -36,21 +36,28 @@ class WriteRpc;
class KuduTable;
-// A write operation operates on a single table and partial row.
-// The KuduWriteOperation class itself allows the batcher to get to the
-// generic information that it needs to process all write operations.
+// A single-row write operation to be sent to a Kudu table.
//
-// On its own, the class does not represent any specific change and thus cannot
-// be constructed independently.
+// This is the abstract base class from which the particular row operations
+// (KuduInsert, KuduUpdate, etc) are derived. These subclasses are instantiated
+// by KuduTable::NewInsert(), etc.
//
-// KuduWriteOperation also holds shared ownership of its KuduTable to allow client's
-// scope to end while the KuduWriteOperation is still alive.
+// The row key, as well as the columns to be inserted or updated are set using
+// the embedded KuduPartialRow object accessible via mutable_row().
+//
+// Typical usage example:
+//
+// KuduInsert* t = table->NewInsert();
+// KUDU_CHECK_OK(t->mutable_row()->SetInt32("key", 1234));
+// KUDU_CHECK_OK(t->mutable_row()->SetStringCopy("foo", "bar"));
+// session->Apply(t);
class KUDU_EXPORT KuduWriteOperation {
public:
enum Type {
INSERT = 1,
UPDATE = 2,
DELETE = 3,
+ UPSERT = 4
};
virtual ~KuduWriteOperation();
@@ -63,6 +70,8 @@ class KUDU_EXPORT KuduWriteOperation {
explicit KuduWriteOperation(const sp::shared_ptr<KuduTable>& table);
virtual Type type() const = 0;
+ // KuduWriteOperation holds shared ownership of its KuduTable to allow the client's
+ // scope to end while the KuduWriteOperation is still alive.
sp::shared_ptr<KuduTable> const table_;
KuduPartialRow row_;
@@ -84,10 +93,10 @@ class KUDU_EXPORT KuduWriteOperation {
DISALLOW_COPY_AND_ASSIGN(KuduWriteOperation);
};
+
// A single row insert to be sent to the cluster.
-// Row operation is defined by what's in the PartialRow instance here.
-// Use mutable_row() to change the row being inserted
-// An insert requires all key columns from the table schema to be defined.
+// An insert requires all key columns to be set, as well as all
+// columns which do not have default values.
class KUDU_EXPORT KuduInsert : public KuduWriteOperation {
public:
virtual ~KuduInsert();
@@ -104,12 +113,28 @@ class KUDU_EXPORT KuduInsert : public KuduWriteOperation {
explicit KuduInsert(const sp::shared_ptr<KuduTable>& table);
};
+// A single row upsert to be sent to the cluster.
+// See KuduInsert for more details.
+class KUDU_EXPORT KuduUpsert : public KuduWriteOperation {
+ public:
+ virtual ~KuduUpsert();
+
+ virtual std::string ToString() const OVERRIDE { return "UPSERT " + row_.ToString(); }
+
+ protected:
+ virtual Type type() const OVERRIDE {
+ return UPSERT;
+ }
+
+ private:
+ friend class KuduTable;
+ explicit KuduUpsert(const sp::shared_ptr<KuduTable>& table);
+};
+
// A single row update to be sent to the cluster.
-// Row operation is defined by what's in the PartialRow instance here.
-// Use mutable_row() to change the row being updated.
// An update requires the key columns and at least one other column
-// in the schema to be defined.
+// in the schema to be set in the embedded PartialRow object.
class KUDU_EXPORT KuduUpdate : public KuduWriteOperation {
public:
virtual ~KuduUpdate();
@@ -128,9 +153,7 @@ class KUDU_EXPORT KuduUpdate : public KuduWriteOperation {
// A single row delete to be sent to the cluster.
-// Row operation is defined by what's in the PartialRow instance here.
-// Use mutable_row() to change the row being deleted
-// A delete requires just the key columns to be defined.
+// A delete requires the key columns to be set in the embedded PartialRow object.
class KUDU_EXPORT KuduDelete : public KuduWriteOperation {
public:
virtual ~KuduDelete();
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/common/row_operations-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_operations-test.cc b/src/kudu/common/row_operations-test.cc
index 4ae3418..0d29e44 100644
--- a/src/kudu/common/row_operations-test.cc
+++ b/src/kudu/common/row_operations-test.cc
@@ -96,7 +96,7 @@ void RowOperationsTest::CheckDecodeDoesntCrash(const Schema& client_schema,
void RowOperationsTest::DoFuzzTest(const Schema& server_schema,
const KuduPartialRow& row,
int n_random_changes) {
- for (int operation = 0; operation < 3; operation++) {
+ for (int operation = 0; operation <= 4; operation++) {
RowOperationsPB pb;
RowOperationsPBEncoder enc(&pb);
@@ -105,12 +105,15 @@ void RowOperationsTest::DoFuzzTest(const Schema& server_schema,
enc.Add(RowOperationsPB::INSERT, row);
break;
case 1:
- enc.Add(RowOperationsPB::UPDATE, row);
+ enc.Add(RowOperationsPB::UPSERT, row);
break;
case 2:
- enc.Add(RowOperationsPB::DELETE, row);
+ enc.Add(RowOperationsPB::UPDATE, row);
break;
case 3:
+ enc.Add(RowOperationsPB::DELETE, row);
+ break;
+ case 4:
enc.Add(RowOperationsPB::SPLIT_ROW, row);
break;
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/common/row_operations.cc
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_operations.cc b/src/kudu/common/row_operations.cc
index 896b4c2..ac6e937 100644
--- a/src/kudu/common/row_operations.cc
+++ b/src/kudu/common/row_operations.cc
@@ -35,6 +35,8 @@ string DecodedRowOperation::ToString(const Schema& schema) const {
switch (type) {
case RowOperationsPB::INSERT:
return "INSERT " + schema.DebugRow(ConstContiguousRow(&schema, row_data));
+ case RowOperationsPB::UPSERT:
+ return "UPSERT " + schema.DebugRow(ConstContiguousRow(&schema, row_data));
case RowOperationsPB::UPDATE:
case RowOperationsPB::DELETE:
return Substitute("MUTATE $0 $1",
@@ -298,9 +300,9 @@ class ClientServerMapping {
};
-Status RowOperationsPBDecoder::DecodeInsert(const uint8_t* prototype_row_storage,
- const ClientServerMapping& mapping,
- DecodedRowOperation* op) {
+Status RowOperationsPBDecoder::DecodeInsertOrUpsert(const uint8_t* prototype_row_storage,
+ const ClientServerMapping& mapping,
+ DecodedRowOperation* op) {
const uint8_t* client_isset_map;
const uint8_t* client_null_map;
@@ -312,8 +314,10 @@ Status RowOperationsPBDecoder::DecodeInsert(const uint8_t* prototype_row_storage
// Allocate a row with the tablet's layout.
uint8_t* tablet_row_storage = reinterpret_cast<uint8_t*>(
- dst_arena_->AllocateBytesAligned(tablet_row_size_, 8));
- if (PREDICT_FALSE(!tablet_row_storage)) {
+ dst_arena_->AllocateBytesAligned(tablet_row_size_, 8));
+ uint8_t* tablet_isset_bitmap = reinterpret_cast<uint8_t*>(
+ dst_arena_->AllocateBytes(BitmapSize(tablet_schema_->num_columns())));
+ if (PREDICT_FALSE(!tablet_row_storage || !tablet_isset_bitmap)) {
return Status::RuntimeError("Out of memory");
}
@@ -335,7 +339,9 @@ Status RowOperationsPBDecoder::DecodeInsert(const uint8_t* prototype_row_storage
DCHECK_GE(tablet_col_idx, 0);
const ColumnSchema& col = tablet_schema_->column(tablet_col_idx);
- if (BitmapTest(client_isset_map, client_col_idx)) {
+ bool isset = BitmapTest(client_isset_map, client_col_idx);
+ BitmapChange(tablet_isset_bitmap, tablet_col_idx, isset);
+ if (isset) {
// If the client provided a value for this column, copy it.
// Copy null-ness, if the server side column is nullable.
@@ -362,6 +368,7 @@ Status RowOperationsPBDecoder::DecodeInsert(const uint8_t* prototype_row_storage
}
op->row_data = tablet_row_storage;
+ op->isset_bitmap = tablet_isset_bitmap;
return Status::OK();
}
@@ -557,7 +564,8 @@ Status RowOperationsPBDecoder::DecodeOperations(vector<DecodedRowOperation>* ops
case RowOperationsPB::UNKNOWN:
return Status::NotSupported("Unknown row operation type");
case RowOperationsPB::INSERT:
- RETURN_NOT_OK(DecodeInsert(prototype_row_storage, mapping, &op));
+ case RowOperationsPB::UPSERT:
+ RETURN_NOT_OK(DecodeInsertOrUpsert(prototype_row_storage, mapping, &op));
break;
case RowOperationsPB::UPDATE:
case RowOperationsPB::DELETE:
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/common/row_operations.h
----------------------------------------------------------------------
diff --git a/src/kudu/common/row_operations.h b/src/kudu/common/row_operations.h
index e7ee722..3bce09b 100644
--- a/src/kudu/common/row_operations.h
+++ b/src/kudu/common/row_operations.h
@@ -51,10 +51,15 @@ class RowOperationsPBEncoder {
struct DecodedRowOperation {
RowOperationsPB::Type type;
- // For INSERT, the whole projected row.
+ // For INSERT or UPSERT, the whole projected row.
// For UPDATE or DELETE, the row key.
const uint8_t* row_data;
+ // For INSERT or UPDATE, a bitmap indicating which of the cells were
+ // explicitly set by the client, versus being filled-in defaults.
+ // A set bit indicates that the client explicitly set the cell.
+ const uint8_t* isset_bitmap;
+
// For UPDATE and DELETE types, the changelist
RowChangeList changelist;
@@ -82,9 +87,9 @@ class RowOperationsPBDecoder {
Status ReadColumn(const ColumnSchema& col, uint8_t* dst);
bool HasNext() const;
- Status DecodeInsert(const uint8_t* prototype_row_storage,
- const ClientServerMapping& mapping,
- DecodedRowOperation* op);
+ Status DecodeInsertOrUpsert(const uint8_t* prototype_row_storage,
+ const ClientServerMapping& mapping,
+ DecodedRowOperation* op);
//------------------------------------------------------------
// Serialization/deserialization support
//------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/common/wire_protocol.proto
----------------------------------------------------------------------
diff --git a/src/kudu/common/wire_protocol.proto b/src/kudu/common/wire_protocol.proto
index 936c8e6..06c6a2b 100644
--- a/src/kudu/common/wire_protocol.proto
+++ b/src/kudu/common/wire_protocol.proto
@@ -138,13 +138,15 @@ message RowwiseRowBlockPB {
optional int32 indirect_data_sidecar = 3;
}
-// A set of operations (INSERT, UPDATE, or DELETE) to apply to a table.
+// A set of operations to apply to a table.
message RowOperationsPB {
enum Type {
UNKNOWN = 0;
INSERT = 1;
UPDATE = 2;
DELETE = 3;
+ UPSERT = 5;
+
// Used when specifying split rows on table creation.
SPLIT_ROW = 4;
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/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 da585cc..e67f03d 100644
--- a/src/kudu/integration-tests/fuzz-itest.cc
+++ b/src/kudu/integration-tests/fuzz-itest.cc
@@ -51,6 +51,7 @@ using std::unique_ptr;
// the fuzz test.
enum TestOpType {
TEST_INSERT,
+ TEST_UPSERT,
TEST_UPDATE,
TEST_DELETE,
TEST_FLUSH_OPS,
@@ -82,13 +83,16 @@ using client::KuduSession;
using client::KuduTable;
using client::KuduTableCreator;
using client::KuduUpdate;
+using client::KuduUpsert;
using client::KuduValue;
+using client::KuduWriteOperation;
using client::sp::shared_ptr;
namespace tablet {
const char* TestOpType_names[] = {
"TEST_INSERT",
+ "TEST_UPSERT",
"TEST_UPDATE",
"TEST_DELETE",
"TEST_FLUSH_OPS",
@@ -105,7 +109,7 @@ struct TestOp {
// The op to run.
TestOpType type;
- // For INSERT/UPDATE/DELETE, the key of the row to be modified. Otherwise, unused.
+ // For INSERT/UPSERT/UPDATE/DELETE, the key of the row to be modified. Otherwise, unused.
int row_key;
string ToString() const {
@@ -189,9 +193,14 @@ class FuzzTest : public KuduTest {
// 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) {
- unique_ptr<KuduInsert> ins(table_->NewInsert());
- KuduPartialRow* row = ins->mutable_row();
+ string InsertOrUpsertRow(int key, int val, TestOpType type) {
+ unique_ptr<KuduWriteOperation> op;
+ if (type == TEST_INSERT) {
+ op.reset(table_->NewInsert());
+ } else {
+ op.reset(table_->NewUpsert());
+ }
+ KuduPartialRow* row = op->mutable_row();
CHECK_OK(row->SetInt32(0, key));
if (val & 1) {
CHECK_OK(row->SetNull(1));
@@ -199,7 +208,7 @@ class FuzzTest : public KuduTest {
CHECK_OK(row->SetInt32(1, val));
}
string ret = row->ToString();
- CHECK_OK(session_->Apply(ins.release()));
+ CHECK_OK(session_->Apply(op.release()));
return ret;
}
@@ -279,6 +288,20 @@ void GenerateTestCase(vector<TestOp>* ops, int len) {
ops_pending = true;
data_in_mrs = true;
break;
+ case TEST_UPSERT:
+ ops->push_back({TEST_UPSERT, row_key});
+ exists[row_key] = true;
+ ops_pending = true;
+ // If the row doesn't currently exist, this will act like an insert
+ // and put it into MRS.
+ if (!exists[row_key]) {
+ data_in_mrs = true;
+ } else if (!data_in_mrs) {
+ // If it does exist, but not in MRS, then this will put data into
+ // a DMS.
+ data_in_dms = true;
+ }
+ break;
case TEST_UPDATE:
if (!exists[row_key]) continue;
ops->push_back({TEST_UPDATE, row_key});
@@ -374,7 +397,8 @@ void FuzzTest::RunFuzzCase(const vector<TestOp>& test_ops,
LOG(INFO) << test_op.ToString();
switch (test_op.type) {
case TEST_INSERT:
- pending_val[test_op.row_key] = InsertRow(test_op.row_key, i++);
+ case TEST_UPSERT:
+ pending_val[test_op.row_key] = InsertOrUpsertRow(test_op.row_key, i++, test_op.type);
break;
case TEST_UPDATE:
for (int j = 0; j < update_multiplier; j++) {
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/local_tablet_writer.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/local_tablet_writer.h b/src/kudu/tablet/local_tablet_writer.h
index 0e94272..7d488e8 100644
--- a/src/kudu/tablet/local_tablet_writer.h
+++ b/src/kudu/tablet/local_tablet_writer.h
@@ -63,6 +63,10 @@ class LocalTabletWriter {
return Write(RowOperationsPB::INSERT, row);
}
+ Status Upsert(const KuduPartialRow& row) {
+ return Write(RowOperationsPB::UPSERT, row);
+ }
+
Status Delete(const KuduPartialRow& row) {
return Write(RowOperationsPB::DELETE, row);
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/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 6d726cc..d1172bb 100644
--- a/src/kudu/tablet/tablet-test-base.h
+++ b/src/kudu/tablet/tablet-test-base.h
@@ -317,14 +317,35 @@ class TabletTestBase : public KuduTabletTest {
int64_t count,
int32_t val,
TimeSeries *ts = NULL) {
+ InsertOrUpsertTestRows(RowOperationsPB::INSERT, first_row, count, val, ts);
+ }
+ // Upserts "count" rows.
+ void UpsertTestRows(int64_t first_row,
+ int64_t count,
+ int32_t val,
+ TimeSeries *ts = NULL) {
+ InsertOrUpsertTestRows(RowOperationsPB::UPSERT, first_row, count, val, ts);
+ }
+
+ void InsertOrUpsertTestRows(RowOperationsPB::Type type,
+ int64_t first_row,
+ int64_t count,
+ int32_t val,
+ TimeSeries *ts = NULL) {
LocalTabletWriter writer(tablet().get(), &client_schema_);
KuduPartialRow row(&client_schema_);
uint64_t inserted_since_last_report = 0;
for (int64_t i = first_row; i < first_row + count; i++) {
setup_.BuildRow(&row, i, val);
- CHECK_OK(writer.Insert(row));
+ if (type == RowOperationsPB::INSERT) {
+ CHECK_OK(writer.Insert(row));
+ } else if (type == RowOperationsPB::UPSERT) {
+ CHECK_OK(writer.Upsert(row));
+ } else {
+ LOG(FATAL) << "bad type: " << type;
+ }
if ((inserted_since_last_report++ > 100) && ts) {
ts->AddValue(static_cast<double>(inserted_since_last_report));
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/tablet-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet-test.cc b/src/kudu/tablet/tablet-test.cc
index c2f13ca..6d33332 100644
--- a/src/kudu/tablet/tablet-test.cc
+++ b/src/kudu/tablet/tablet-test.cc
@@ -567,6 +567,28 @@ TYPED_TEST(TestTablet, TestInsertsPersist) {
// TODO: add some more data, re-flush
}
+TYPED_TEST(TestTablet, TestUpsert) {
+ vector<string> rows;
+ this->UpsertTestRows(0, 1, 1000);
+
+ // UPSERT a row that is in MRS.
+ this->UpsertTestRows(0, 1, 1001);
+
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{ this->setup_.FormatDebugRow(0, 1001, false) }, rows);
+
+ // Flush it.
+ ASSERT_OK(this->tablet()->Flush());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{ this->setup_.FormatDebugRow(0, 1001, false) }, rows);
+
+ // UPSERT a row that is in DRS.
+ this->UpsertTestRows(0, 1, 1002);
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{ this->setup_.FormatDebugRow(0, 1002, false) }, rows);
+}
+
+
// Test that when a row has been updated many times, it always yields
// the most recent value.
TYPED_TEST(TestTablet, TestMultipleUpdates) {
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/tablet.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index 0810744..ad5067b 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -384,45 +384,92 @@ void Tablet::StartTransaction(WriteTransactionState* tx_state) {
tx_state->SetMvccTxAndTimestamp(std::move(mvcc_tx));
}
-Status Tablet::InsertUnlocked(WriteTransactionState *tx_state,
- RowOp* insert,
- ProbeStats* stats) {
+Status Tablet::InsertOrUpsertUnlocked(WriteTransactionState *tx_state,
+ RowOp* op,
+ ProbeStats* stats) {
+ const bool is_upsert = op->decoded_op.type == RowOperationsPB::UPSERT;
const TabletComponents* comps = DCHECK_NOTNULL(tx_state->tablet_components());
// First, ensure that it is a unique key by checking all the open RowSets.
vector<RowSet *> to_check;
- comps->rowsets->FindRowSetsWithKeyInRange(insert->key_probe->encoded_key_slice(),
+ comps->rowsets->FindRowSetsWithKeyInRange(op->key_probe->encoded_key_slice(),
&to_check);
- for (const RowSet *rowset : to_check) {
+ for (RowSet *rowset : to_check) {
bool present = false;
- RETURN_NOT_OK(rowset->CheckRowPresent(*insert->key_probe, &present, stats));
- if (PREDICT_FALSE(present)) {
+ RETURN_NOT_OK(rowset->CheckRowPresent(*op->key_probe, &present, stats));
+ if (present) {
+ if (is_upsert) {
+ return ApplyUpsertAsUpdate(tx_state, op, rowset, stats);
+ }
Status s = Status::AlreadyPresent("key already present");
if (metrics_) {
metrics_->insertions_failed_dup_key->Increment();
}
- insert->SetFailed(s);
+ op->SetFailed(s);
return s;
}
}
Timestamp ts = tx_state->timestamp();
- ConstContiguousRow row(schema(), insert->decoded_op.row_data);
+ ConstContiguousRow row(schema(), op->decoded_op.row_data);
// TODO: the Insert() call below will re-encode the key, which is a
// waste. Should pass through the KeyProbe structure perhaps.
- // Now try to insert into memrowset. The memrowset itself will return
- // AlreadyPresent if it has already been inserted there.
+ // Now try to op into memrowset. The memrowset itself will return
+ // AlreadyPresent if it has already been oped there.
Status s = comps->memrowset->Insert(ts, row, tx_state->op_id());
- if (PREDICT_TRUE(s.ok())) {
- insert->SetInsertSucceeded(comps->memrowset->mrs_id());
+ if (s.ok()) {
+ op->SetInsertSucceeded(comps->memrowset->mrs_id());
} else {
- if (s.IsAlreadyPresent() && metrics_) {
- metrics_->insertions_failed_dup_key->Increment();
+ if (s.IsAlreadyPresent()) {
+ if (is_upsert) {
+ return ApplyUpsertAsUpdate(tx_state, op, comps->memrowset.get(), stats);
+ }
+ if (metrics_) {
+ metrics_->insertions_failed_dup_key->Increment();
+ }
}
- insert->SetFailed(s);
+ op->SetFailed(s);
+ }
+ return s;
+}
+
+Status Tablet::ApplyUpsertAsUpdate(WriteTransactionState* tx_state,
+ RowOp* upsert,
+ RowSet* rowset,
+ ProbeStats* stats) {
+ const auto* schema = this->schema();
+ ConstContiguousRow row(schema, upsert->decoded_op.row_data);
+ faststring buf;
+ RowChangeListEncoder enc(&buf);
+ for (int i = 0; i < schema->num_columns(); i++) {
+ if (schema->is_key_column(i)) continue;
+
+ // If the user didn't explicitly set this column in the UPSERT, then we should
+ // not turn it into an UPDATE. This prevents the UPSERT from updating
+ // values back to their defaults when unset.
+ if (!BitmapTest(upsert->decoded_op.isset_bitmap, i)) continue;
+ const auto& c = schema->column(i);
+ const void* val = c.is_nullable() ? row.nullable_cell_ptr(i) : row.cell_ptr(i);
+ enc.AddColumnUpdate(c, schema->column_id(i), val);
+ }
+
+ RowChangeList rcl = enc.as_changelist();
+
+ gscoped_ptr<OperationResultPB> result(new OperationResultPB());
+ Status s = rowset->MutateRow(tx_state->timestamp(),
+ *upsert->key_probe,
+ rcl,
+ tx_state->op_id(),
+ stats,
+ result.get());
+ CHECK(!s.IsNotFound());
+ if (s.ok()) {
+ upsert->SetMutateSucceeded(std::move(result));
+ } else {
+ upsert->SetFailed(s);
}
return s;
}
@@ -570,7 +617,8 @@ void Tablet::ApplyRowOperation(WriteTransactionState* tx_state,
switch (row_op->decoded_op.type) {
case RowOperationsPB::INSERT:
- ignore_result(InsertUnlocked(tx_state, row_op, stats));
+ case RowOperationsPB::UPSERT:
+ ignore_result(InsertOrUpsertUnlocked(tx_state, row_op, stats));
return;
case RowOperationsPB::UPDATE:
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/tablet.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index c46aa22..1e12698 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -383,20 +383,28 @@ class Tablet {
Status FlushUnlocked();
- // A version of Insert that does not acquire locks and instead assumes that
- // they were already acquired. Requires that handles for the relevant locks
- // and MVCC transaction are present in the transaction state.
- Status InsertUnlocked(WriteTransactionState *tx_state,
- RowOp* insert,
- ProbeStats* stats);
-
- // A version of MutateRow that does not acquire locks and instead assumes
- // they were already acquired. Requires that handles for the relevant locks
- // and MVCC transaction are present in the transaction state.
+
+ // Perform an INSERT or UPSERT operation, assuming that the transaction is already in
+ // prepared state. This state ensures that:
+ // - the row lock is acquired
+ // - the tablet components have been acquired
+ // - the operation has been decoded
+ Status InsertOrUpsertUnlocked(WriteTransactionState *tx_state,
+ RowOp* insert,
+ ProbeStats* stats);
+
+ // Same as above, but for UPDATE.
Status MutateRowUnlocked(WriteTransactionState *tx_state,
RowOp* mutate,
ProbeStats* stats);
+ // In the case of an UPSERT against a duplicate row, converts the UPSERT
+ // into an internal UPDATE operation and performs it.
+ Status ApplyUpsertAsUpdate(WriteTransactionState *tx_state,
+ RowOp* upsert,
+ RowSet* rowset,
+ ProbeStats* stats);
+
// Return the list of RowSets that need to be consulted when processing the
// given mutation.
static std::vector<RowSet*> FindRowSetsToCheck(RowOp* mutate,
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/tablet_bootstrap.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_bootstrap.cc b/src/kudu/tablet/tablet_bootstrap.cc
index f63143b..ab4422a 100644
--- a/src/kudu/tablet/tablet_bootstrap.cc
+++ b/src/kudu/tablet/tablet_bootstrap.cc
@@ -1324,7 +1324,9 @@ Status TabletBootstrap::ApplyOperations(WriteTransactionState* tx_state,
// Increment the seen/ignored stats.
switch (op->decoded_op.type) {
- case RowOperationsPB::INSERT: {
+ case RowOperationsPB::INSERT:
+ case RowOperationsPB::UPSERT: {
+ // TODO: should we have a separate counter for upserts?
stats_.inserts_seen++;
if (op->has_result()) {
stats_.inserts_ignored++;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/tablet_metrics.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_metrics.cc b/src/kudu/tablet/tablet_metrics.cc
index 0b8b196..7b776a4 100644
--- a/src/kudu/tablet/tablet_metrics.cc
+++ b/src/kudu/tablet/tablet_metrics.cc
@@ -29,6 +29,9 @@
METRIC_DEFINE_counter(tablet, rows_inserted, "Rows Inserted",
kudu::MetricUnit::kRows,
"Number of rows inserted into this tablet since service start");
+METRIC_DEFINE_counter(tablet, rows_upserted, "Rows Upserted",
+ kudu::MetricUnit::kRows,
+ "Number of rows upserted into this tablet since service start");
METRIC_DEFINE_counter(tablet, rows_updated, "Rows Updated",
kudu::MetricUnit::kRows,
"Number of row update operations performed on this tablet since service start");
@@ -220,6 +223,7 @@ namespace tablet {
#define GINIT(x) x(METRIC_##x.Instantiate(entity, 0))
TabletMetrics::TabletMetrics(const scoped_refptr<MetricEntity>& entity)
: MINIT(rows_inserted),
+ MINIT(rows_upserted),
MINIT(rows_updated),
MINIT(rows_deleted),
MINIT(insertions_failed_dup_key),
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/tablet_metrics.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/tablet_metrics.h b/src/kudu/tablet/tablet_metrics.h
index 0d20a51..84a7185 100644
--- a/src/kudu/tablet/tablet_metrics.h
+++ b/src/kudu/tablet/tablet_metrics.h
@@ -47,6 +47,7 @@ struct TabletMetrics {
// Operation rates
scoped_refptr<Counter> rows_inserted;
+ scoped_refptr<Counter> rows_upserted;
scoped_refptr<Counter> rows_updated;
scoped_refptr<Counter> rows_deleted;
scoped_refptr<Counter> insertions_failed_dup_key;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/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 8295ff9..ae9822c 100644
--- a/src/kudu/tablet/tablet_random_access-test.cc
+++ b/src/kudu/tablet/tablet_random_access-test.cc
@@ -91,13 +91,23 @@ class TestRandomAccess : public KuduTabletTest {
for (int i = 0; i < 3; i++) {
int new_val = rand();
if (cur_val.empty()) {
- // If there is no row, then insert one.
- cur_val = InsertRow(key, new_val, &pending);
+ // 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);
+ }
} else {
if (new_val % (FLAGS_update_delete_ratio + 1) == 0) {
cur_val = DeleteRow(key, &pending);
} else {
- cur_val = MutateRow(key, new_val, &pending);
+ // 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);
+ } else {
+ cur_val = UpsertRow(key, new_val, &pending);
+ }
}
}
}
@@ -144,6 +154,15 @@ 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);
+ }
+
+ 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) {
@@ -152,7 +171,7 @@ class TestRandomAccess : public KuduTabletTest {
CHECK_OK(row->SetInt32(1, val));
}
string ret = row->ToString();
- ops->push_back(LocalTabletWriter::Op(RowOperationsPB::INSERT, row.release()));
+ ops->push_back(LocalTabletWriter::Op(type, row.release()));
return ret;
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/transactions/transaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/transactions/transaction.cc b/src/kudu/tablet/transactions/transaction.cc
index 5a0f737..b8681a1 100644
--- a/src/kudu/tablet/transactions/transaction.cc
+++ b/src/kudu/tablet/transactions/transaction.cc
@@ -71,6 +71,7 @@ TransactionCompletionCallback::~TransactionCompletionCallback() {}
TransactionMetrics::TransactionMetrics()
: successful_inserts(0),
+ successful_upserts(0),
successful_updates(0),
successful_deletes(0),
commit_wait_duration_usec(0) {
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/transactions/transaction.h
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/transactions/transaction.h b/src/kudu/tablet/transactions/transaction.h
index 5774efb..dc49fd3 100644
--- a/src/kudu/tablet/transactions/transaction.h
+++ b/src/kudu/tablet/transactions/transaction.h
@@ -40,6 +40,7 @@ struct TransactionMetrics {
TransactionMetrics();
void Reset();
int successful_inserts;
+ int successful_upserts;
int successful_updates;
int successful_deletes;
uint64_t commit_wait_duration_usec;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/56c43158/src/kudu/tablet/transactions/write_transaction.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/transactions/write_transaction.cc b/src/kudu/tablet/transactions/write_transaction.cc
index 67511fa..0959350 100644
--- a/src/kudu/tablet/transactions/write_transaction.cc
+++ b/src/kudu/tablet/transactions/write_transaction.cc
@@ -166,6 +166,7 @@ void WriteTransaction::Finish(TransactionResult result) {
// TODO: should we change this so it's actually incremented by the
// Tablet code itself instead of this wrapper code?
metrics->rows_inserted->IncrementBy(state_->metrics().successful_inserts);
+ metrics->rows_upserted->IncrementBy(state_->metrics().successful_upserts);
metrics->rows_updated->IncrementBy(state_->metrics().successful_updates);
metrics->rows_deleted->IncrementBy(state_->metrics().successful_deletes);
@@ -289,6 +290,9 @@ void WriteTransactionState::UpdateMetricsForOp(const RowOp& op) {
case RowOperationsPB::INSERT:
tx_metrics_.successful_inserts++;
break;
+ case RowOperationsPB::UPSERT:
+ tx_metrics_.successful_upserts++;
+ break;
case RowOperationsPB::UPDATE:
tx_metrics_.successful_updates++;
break;
[2/2] incubator-kudu git commit: [java client] Add Upsert
Posted by to...@apache.org.
[java client] Add Upsert
Change-Id: I51112981cc82146802e8484000a6b74dd724116e
Reviewed-on: http://gerrit.cloudera.org:8080/3123
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: Kudu Jenkins
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/591c9ccb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/591c9ccb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/591c9ccb
Branch: refs/heads/master
Commit: 591c9ccb2a08bcfa4cf7cce32fc60b05b0d2ca86
Parents: 56c4315
Author: Jean-Daniel Cryans <jd...@apache.org>
Authored: Wed May 18 12:34:59 2016 -0700
Committer: Todd Lipcon <to...@apache.org>
Committed: Thu May 19 06:26:40 2016 +0000
----------------------------------------------------------------------
.../main/java/org/kududb/client/KuduTable.java | 8 ++++
.../main/java/org/kududb/client/Operation.java | 3 +-
.../src/main/java/org/kududb/client/Upsert.java | 37 ++++++++++++++++++
.../java/org/kududb/client/TestKuduSession.java | 40 ++++++++++++++++++++
4 files changed, 87 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/591c9ccb/java/kudu-client/src/main/java/org/kududb/client/KuduTable.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/kududb/client/KuduTable.java b/java/kudu-client/src/main/java/org/kududb/client/KuduTable.java
index b0dd4e7..aa9ec29 100644
--- a/java/kudu-client/src/main/java/org/kududb/client/KuduTable.java
+++ b/java/kudu-client/src/main/java/org/kududb/client/KuduTable.java
@@ -129,6 +129,14 @@ public class KuduTable {
}
/**
+ * Get a new upsert configured with this table's schema. The returned object should not be reused.
+ * @return an upsert with this table's schema
+ */
+ public Upsert newUpsert() {
+ return new Upsert(this);
+ }
+
+ /**
* Get all the tablets for this table. This may query the master multiple times if there
* are a lot of tablets.
* @param deadline deadline in milliseconds for this method to finish
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/591c9ccb/java/kudu-client/src/main/java/org/kududb/client/Operation.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/kududb/client/Operation.java b/java/kudu-client/src/main/java/org/kududb/client/Operation.java
index 417851d..91cadcf 100644
--- a/java/kudu-client/src/main/java/org/kududb/client/Operation.java
+++ b/java/kudu-client/src/main/java/org/kududb/client/Operation.java
@@ -60,7 +60,8 @@ public abstract class Operation extends KuduRpc<OperationResponse> implements Ku
INSERT((byte)RowOperationsPB.Type.INSERT.getNumber()),
UPDATE((byte)RowOperationsPB.Type.UPDATE.getNumber()),
DELETE((byte)RowOperationsPB.Type.DELETE.getNumber()),
- SPLIT_ROWS((byte)RowOperationsPB.Type.SPLIT_ROW.getNumber());
+ SPLIT_ROWS((byte)RowOperationsPB.Type.SPLIT_ROW.getNumber()),
+ UPSERT((byte)RowOperationsPB.Type.UPSERT.getNumber());
ChangeType(byte encodedByte) {
this.encodedByte = encodedByte;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/591c9ccb/java/kudu-client/src/main/java/org/kududb/client/Upsert.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/main/java/org/kududb/client/Upsert.java b/java/kudu-client/src/main/java/org/kududb/client/Upsert.java
new file mode 100644
index 0000000..4ba2635
--- /dev/null
+++ b/java/kudu-client/src/main/java/org/kududb/client/Upsert.java
@@ -0,0 +1,37 @@
+// 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.
+package org.kududb.client;
+
+import org.kududb.annotations.InterfaceAudience;
+import org.kududb.annotations.InterfaceStability;
+
+/**
+ * Represents a single row upsert. Instances of this class should not be reused.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class Upsert extends Operation {
+
+ Upsert(KuduTable table) {
+ super(table);
+ }
+
+ @Override
+ ChangeType getChangeType() {
+ return ChangeType.UPSERT;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/591c9ccb/java/kudu-client/src/test/java/org/kududb/client/TestKuduSession.java
----------------------------------------------------------------------
diff --git a/java/kudu-client/src/test/java/org/kududb/client/TestKuduSession.java b/java/kudu-client/src/test/java/org/kududb/client/TestKuduSession.java
index 9ac856a..4c07421 100644
--- a/java/kudu-client/src/test/java/org/kududb/client/TestKuduSession.java
+++ b/java/kudu-client/src/test/java/org/kududb/client/TestKuduSession.java
@@ -189,7 +189,47 @@ public class TestKuduSession extends BaseKuduTest {
}
}
+ @Test(timeout = 10000)
+ public void testUpsert() throws Exception {
+ String tableName = TABLE_NAME_PREFIX + "-Upsert";
+ table = createTable(tableName, basicSchema, null);
+ KuduSession session = syncClient.newSession();
+
+ // Test an Upsert that acts as an Insert.
+ assertFalse(session.apply(createUpsert(1, 1, false)).hasRowError());
+
+ List<String> rowStrings = scanTableToStrings(table);
+ assertEquals(1, rowStrings.size());
+ assertEquals(
+ "INT32 key=1, INT32 column1_i=1, INT32 column2_i=3, " +
+ "STRING column3_s=a string, BOOL column4_b=true",
+ rowStrings.get(0));
+
+ // Test an Upsert that acts as an Update.
+ assertFalse(session.apply(createUpsert(1, 2, false)).hasRowError());
+ rowStrings = scanTableToStrings(table);
+ assertEquals(
+ "INT32 key=1, INT32 column1_i=2, INT32 column2_i=3, " +
+ "STRING column3_s=a string, BOOL column4_b=true",
+ rowStrings.get(0));
+ }
+
private Insert createInsert(int key) {
return createBasicSchemaInsert(table, key);
}
+
+ private Upsert createUpsert(int key, int secondVal, boolean hasNull) {
+ Upsert upsert = table.newUpsert();
+ PartialRow row = upsert.getRow();
+ row.addInt(0, key);
+ row.addInt(1, secondVal);
+ row.addInt(2, 3);
+ if (hasNull) {
+ row.setNull(3);
+ } else {
+ row.addString(3, "a string");
+ }
+ row.addBoolean(4, true);
+ return upsert;
+ }
}