Merge branch 'feature/dt-messages' into develop

Conflicts:
	riak_pb/__init__.py

Makefile

@@ -28,6 +28,35 @@ erl_clean:
 erl_test: erl_compile
 	@./rebar eunit skip_deps=true
 
+REPO = riak_pb
+APPS = kernel stdlib sasl erts ssl tools os_mon runtime_tools crypto inets \
+	xmerl webtool snmp public_key mnesia eunit syntax_tools compiler
+COMBO_PLT = $(HOME)/.$(REPO)_combo_dialyzer_plt
+
+check_plt: erl_deps erl_compile
+	dialyzer --check_plt --plt $(COMBO_PLT) --apps $(APPS) \
+		deps/*/ebin
+
+build_plt: erl_deps erl_compile
+	dialyzer --build_plt --output_plt $(COMBO_PLT) --apps $(APPS) \
+		deps/*/ebin
+
+dialyzer: erl_deps erl_compile
+	@echo
+	@echo Use "'make check_plt'" to check PLT prior to using this target.
+	@echo Use "'make build_plt'" to build PLT prior to using this target.
+	@echo
+	@sleep 1
+	dialyzer -Wno_return --plt $(COMBO_PLT) deps/*/ebin
+
+cleanplt:
+	@echo
+	@echo "Are you sure?  It could take a long time to rebuild."
+	@echo Deleting $(COMBO_PLT) in 5 seconds.
+	@echo
+	sleep 5
+	rm $(COMBO_PLT)
+
 # Python specific build steps
 python_compile:
 	@echo "==> Python (compile)"

riak_pb/__init__.py

@@ -1,4 +1,5 @@
 from riak_pb2 import *
 from riak_kv_pb2 import *
 from riak_search_pb2 import *
+from riak_dt_pb2 import *
 from riak_yokozuna_pb2 import *

src/riak_dt.proto

@@ -0,0 +1,255 @@
+/* -------------------------------------------------------------------
+**
+** riak_dt.proto: Protocol buffers for Riak data structures/types
+**
+** Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
+**
+** This file is provided 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.
+**
+** -------------------------------------------------------------------
+*/
+
+/*
+** Revision: 2.0
+*/
+
+// Java package specifiers
+option java_package = "com.basho.riak.protobuf";
+option java_outer_classname = "RiakDtPB";
+
+/*
+ * =============== DATA STRUCTURES =================
+ */
+
+/*
+ * Field names in maps are composed of a binary identifier and a type.
+ * This is so that two clients can create fields with the same name
+ * but different types, and they converge independently.
+ */
+message MapField {
+    /*
+     * The types that can be stored in a map are limited to counters,
+     * sets, registers, flags, and maps.
+     */
+    enum MapFieldType {
+        COUNTER  = 1;
+        SET      = 2;
+        REGISTER = 3;
+        FLAG     = 4;
+        MAP      = 5;
+    }
+
+    required bytes        name = 1;
+    required MapFieldType type = 2;
+}
+
+
+/*
+ * An entry in a map is a pair of a field-name and value. The type
+ * defined in the field determines which value type is expected.
+ */
+message MapEntry {
+    required MapField field = 1;
+    optional sint64   counter_value  = 2;
+    repeated bytes    set_value      = 3;
+    optional bytes    register_value = 4;
+    optional bool     flag_value     = 5;
+    repeated MapEntry map_value      = 6;
+}
+
+/*
+ * =============== FETCH =================
+ */
+
+/*
+ * The equivalent of KV's "RpbGetReq", results in a DtFetchResp. The
+ * request-time options are limited to ones that are relevant to
+ * structured data-types.
+ */
+message DtFetchReq {
+    // The identifier: bucket, key and bucket-type
+    required bytes bucket = 1;
+    required bytes key    = 2;
+    required bytes type  = 3;
+
+    // Request options
+    optional uint32 r             =  4;
+    optional uint32 pr            =  5;
+    optional bool   basic_quorum  =  6;
+    optional bool   notfound_ok   =  7;
+    optional uint32 timeout       =  8;
+    optional bool   sloppy_quorum =  9;  // Experimental, may change/disappear
+    optional uint32 n_val         = 10;  // Experimental, may change/disappear
+
+    // For read-only requests or context-free operations, you can set
+    // this to false to reduce the size of the response payload.
+    optional bool include_context = 11 [default=true];
+}
+
+
+/*
+ * The value of the fetched data type. If present in the response,
+ * then empty values (sets, maps) should be treated as such.
+ */
+message DtValue {
+    optional sint64   counter_value = 1;
+    repeated bytes    set_value     = 2;
+    repeated MapEntry map_value     = 3;
+}
+
+
+/*
+ * The response to a "Fetch" request. If the `include_context` option
+ * is specified, an opaque "context" value will be returned along with
+ * the user-friendly data. When sending an "Update" request, the
+ * client should send this context as well, similar to how one would
+ * send a vclock for KV updates. The `type` field indicates which
+ * value type to expect. When the `value` field is missing from the
+ * message, the client should interpret it as a "not found".
+ */
+message DtFetchResp {
+    enum DataType {
+        COUNTER = 1;
+        SET     = 2;
+        MAP     = 3;
+    }
+
+    optional bytes    context = 1;
+    required DataType type    = 2;
+    optional DtValue  value   = 3;
+}
+
+/*
+ * =============== UPDATE =================
+ */
+
+/*
+ * An operation to update a Counter, either on its own or inside a
+ * Map. The `increment` field can be positive or negative. When absent,
+ * the meaning is an increment by 1.
+ */
+message CounterOp {
+    optional sint64 increment = 1;
+}
+
+/*
+ * An operation to update a Set, either on its own or inside a Map.
+ * Set members are opaque binary values, you can only add or remove
+ * them from a Set.
+ */
+message SetOp {
+    repeated bytes adds    = 1;
+    repeated bytes removes = 2;
+}
+
+/*
+ * An operation to be applied to a value stored in a Map -- the
+ * contents of an UPDATE operation. The operation field that is
+ * present depends on the type of the field to which it is applied.
+ */
+message MapUpdate {
+    /*
+     * Flags only exist inside Maps and can only be enabled or
+     * disabled, and there are no arguments to the operations.
+     */
+    enum FlagOp {
+        ENABLE  = 1;
+        DISABLE = 2;
+    }
+
+    required MapField  field       = 1;
+
+    optional CounterOp counter_op  = 2;
+    optional SetOp     set_op      = 3;
+
+    /*
+     * There is only one operation on a register, which is to set its
+     * value, therefore the "operation" is the new value.
+     */
+    optional bytes     register_op = 4;
+    optional FlagOp    flag_op     = 5;
+    optional MapOp     map_op      = 6;
+}
+
+/*
+ * An operation to update a Map. All operations apply to individual
+ * fields in the Map.
+ */
+message MapOp {
+    /*
+     * ADD creates a new, "empty" value under a field. REMOVE removes
+     * a field and value from the Map. UPDATE applies type-specific
+     * operations to the values stored in the Map.
+     */
+    repeated MapField  adds    = 1;
+    repeated MapField  removes = 2;
+    repeated MapUpdate updates = 3;
+}
+
+/*
+ * A "union" type for update operations. The included operation
+ * depends on the datatype being updated.
+ */
+message DtOp {
+    optional CounterOp counter_op = 1;
+    optional SetOp     set_op     = 2;
+    optional MapOp     map_op     = 3;
+}
+
+/*
+ * The equivalent of KV's "RpbPutReq", results in an empty response or
+ * "DtUpdateResp" if `return_body` is specified, or the key is
+ * assigned by the server. The request-time options are limited to
+ * ones that are relevant to structured data-types.
+ */
+message DtUpdateReq {
+    // The identifier
+    required bytes bucket = 1;
+    optional bytes key    = 2; // missing key results in server-assigned key, like KV
+    required bytes type   = 3; // bucket type, not data-type (but the data-type is constrained per bucket-type)
+
+    // Opaque update-context
+    optional bytes context = 4;
+
+    // The operations
+    required DtOp  op = 5;
+
+    // Request options
+    optional uint32 w               =  6;
+    optional uint32 dw              =  7;
+    optional uint32 pw              =  8;
+    optional bool   return_body     =  9 [default=false];
+    optional uint32 timeout         = 10;
+    optional bool   sloppy_quorum   = 11;  // Experimental, may change/disappear
+    optional uint32 n_val           = 12;  // Experimental, may change/disappear
+    optional bool   include_context = 13 [default=true]; // When return_body is true, should the context be returned too?
+}
+
+
+/*
+ * The equivalent of KV's "RpbPutResp", contains the assigned key if
+ * it was assigned by the server, and the resulting value and context
+ * if return_body was set.
+ */
+message DtUpdateResp {
+    // The key, if assigned by the server
+    optional bytes    key           = 1;
+
+    // The opaque update context and value, if return_body was set.
+    optional bytes    context       = 2;
+    optional sint64   counter_value = 3;
+    repeated bytes    set_value     = 4;
+    repeated MapEntry map_value     = 5;
+}

src/riak_pb_codec.erl

@@ -135,6 +135,10 @@ msg_type(57) -> rpbyokozunaindexdeletereq;
 msg_type(58) -> rpbyokozunaschemagetreq;
 msg_type(59) -> rpbyokozunaschemagetresp;
 msg_type(60) -> rpbyokozunaschemaputreq;
+msg_type(80) -> dtfetchreq;
+msg_type(81) -> dtfetchresp;
+msg_type(82) -> dtupdatereq;
+msg_type(83) -> dtupdateresp;
 msg_type(_) -> undefined.
 
 %% @doc Converts a symbolic message name into a message code. Replaces
@@ -186,7 +190,11 @@ msg_code(rpbyokozunaindexputreq)    -> 56;
 msg_code(rpbyokozunaindexdeletereq) -> 57;
 msg_code(rpbyokozunaschemagetreq)   -> 58;
 msg_code(rpbyokozunaschemagetresp)  -> 59;
-msg_code(rpbyokozunaschemaputreq)   -> 60.
+msg_code(rpbyokozunaschemaputreq)   -> 60;
+msg_code(dtfetchreq)             -> 80;
+msg_code(dtfetchresp)            -> 81;
+msg_code(dtupdatereq)            -> 82;
+msg_code(dtupdateresp)           -> 83.
 
 %% @doc Selects the appropriate PB decoder for a message code.
 -spec decoder_for(pos_integer()) -> module().
@@ -206,7 +214,9 @@ decoder_for(N) when N >= 3, N < 7;
 decoder_for(N) when N >= 27, N =< 28 ->
     riak_search_pb;
 decoder_for(N) when N >= 54, N =< 60 ->
-    riak_yokozuna_pb.
+    riak_yokozuna_pb;
+decoder_for(N) when N >= 80, N =< 83 ->
+    riak_dt_pb.
 
 %% @doc Selects the appropriate PB encoder for a given message name.
 -spec encoder_for(atom()) -> module().