Skip to content

Commit 3b515d2

Browse files
mlucymlucy
committed
Merge branch '1571_json_protocol' into next
2 parents f85414a + 912725d commit 3b515d2

19 files changed

Lines changed: 153 additions & 80 deletions

File tree

‎drivers/javascript/net.coffee‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,9 @@ aropt = util.aropt
1515

1616
deconstructDatum = (datum, opts) ->
1717
pb.DatumTypeSwitch(datum, {
18-
"R_NULL": =>
18+
"R_JSON": =>
19+
JSON.parse(datum.r_str)
20+
,"R_NULL": =>
1921
null
2022
,"R_BOOL": =>
2123
datum.r_bool
@@ -337,6 +339,7 @@ class Connection extends events.EventEmitter
337339
@_sendQuery(query)
338340

339341
_sendQuery: (query) ->
342+
query.accepts_r_json = true
340343

341344
# Serialize protobuf
342345
data = pb.SerializeQuery(query)

‎drivers/python/rethinkdb/ast.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -559,7 +559,9 @@ def compose(self, args, optargs):
559559
@staticmethod
560560
def deconstruct(datum, time_format='native'):
561561
d_type = datum.type
562-
if d_type == p.Datum.R_OBJECT:
562+
if d_type == p.Datum.R_JSON:
563+
return py_json.loads(datum.r_str)
564+
elif d_type == p.Datum.R_OBJECT:
563565
obj = { }
564566
for pair in datum.r_object:
565567
obj[pair.key] = Datum.deconstruct(pair.val, time_format)

‎drivers/python/rethinkdb/net.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -288,6 +288,8 @@ def _send_query(self, query, term, opts={}, async=False):
288288
if not self.socket:
289289
raise RqlDriverError("Connection is closed.")
290290

291+
query.accepts_r_json = True
292+
291293
# Send protobuf
292294
query_protobuf = query.SerializeToString()
293295
query_header = struct.pack("<L", len(query_protobuf))
@@ -308,7 +310,7 @@ def _send_query(self, query, term, opts={}, async=False):
308310
time_format = opts['time_format']
309311

310312
# Sequence responses
311-
elif response.type == p.Response.SUCCESS_PARTIAL or response.type == p.Response.SUCCESS_SEQUENCE:
313+
if response.type == p.Response.SUCCESS_PARTIAL or response.type == p.Response.SUCCESS_SEQUENCE:
312314
value = Cursor(self, query, term, opts)
313315
value._extend(response)
314316

‎drivers/ruby/lib/net.rb‎

Lines changed: 13 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,14 @@
66
# $f = File.open("fuzz_seed.rb", "w")
77

88
module RethinkDB
9+
def self.new_query(type, token)
10+
q = Query.new
11+
q.type = type
12+
q.accepts_r_json = true
13+
q.token = token
14+
return q
15+
end
16+
917
module Faux_Abort
1018
class Abort
1119
end
@@ -70,9 +78,7 @@ def each (&block) # :nodoc:
7078
while true
7179
@results.each(&block)
7280
return self if !@more
73-
q = Query.new
74-
q.type = Query::QueryType::CONTINUE
75-
q.token = @token
81+
q = RethinkDB::new_query(Query::QueryType::CONTINUE, @token)
7682
res = @conn.run_internal q
7783
@results = Shim.response_to_native(res, @msg, @opts)
7884
if res.type == Response::ResponseType::SUCCESS_SEQUENCE
@@ -117,10 +123,8 @@ def run_internal(q, noreply=false)
117123
def run(msg, opts)
118124
reconnect(:noreply_wait => false) if @auto_reconnect && (!@socket || !@listener)
119125
raise RqlRuntimeError, "Error: Connection Closed." if !@socket || !@listener
120-
q = Query.new
121-
q.type = Query::QueryType::START
126+
q = RethinkDB::new_query(Query::QueryType::START, @@token_cnt += 1)
122127
q.query = msg
123-
q.token = @@token_cnt += 1
124128

125129
all_opts = @default_opts.merge(opts)
126130
if all_opts.keys.include?(:noreply)
@@ -213,7 +217,7 @@ def reconnect(opts={})
213217
(opts.keys - [:noreply_wait]).to_s
214218
end
215219
opts[:noreply_wait] = true if not opts.keys.include?(:noreply_wait)
216-
220+
217221
self.noreply_wait() if opts[:noreply_wait]
218222
@socket.close if @socket
219223
@socket = TCPSocket.open(@host, @port)
@@ -232,7 +236,7 @@ def close(opts={})
232236
(opts.keys - [:noreply_wait]).to_s
233237
end
234238
opts[:noreply_wait] = true if not opts.keys.include?(:noreply_wait)
235-
239+
236240
self.noreply_wait() if opts[:noreply_wait]
237241
@listener.terminate if @listener
238242
@listener = nil
@@ -242,9 +246,7 @@ def close(opts={})
242246

243247
def noreply_wait
244248
raise RqlRuntimeError, "Error: Connection Closed." if !@socket || !@listener
245-
q = Query.new
246-
q.type = Query::QueryType::NOREPLY_WAIT
247-
q.token = @@token_cnt += 1
249+
q = RethinkDB::new_query(Query::QueryType::NOREPLY_WAIT, @@token_cnt += 1)
248250
res = run_internal(q)
249251
if res.type != Response::ResponseType::WAIT_COMPLETE
250252
raise RqlRuntimeError, "Unexpected response to noreply_wait: " + PP.pp(res, "")

‎drivers/ruby/lib/shim.rb‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,9 @@ def self.datum_to_native(d, opts)
2121
else
2222
obj
2323
end
24-
else raise RqlRuntimeError, "Unimplemented."
24+
when dt::R_JSON then
25+
JSON.parse("[" + d.r_str + "]")[0]
26+
else raise RqlRuntimeError, "#{dt} Unimplemented."
2527
end
2628
end
2729

‎src/rdb_protocol/batching.cc‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,9 @@ batchspec_t::batchspec_t(
1818
size_left(size),
1919
end_time(_batch_type == batch_type_t::NORMAL
2020
? _end_time
21-
: std::numeric_limits<decltype(batchspec_t().end_time)>::max()) { }
21+
: std::numeric_limits<decltype(batchspec_t().end_time)>::max()) {
22+
r_sanity_check(els_left >= 1);
23+
}
2224

2325
batchspec_t batchspec_t::user(batch_type_t batch_type,
2426
const counted_t<const datum_t> &conf) {
@@ -50,6 +52,15 @@ batchspec_t batchspec_t::with_new_batch_type(batch_type_t new_batch_type) const
5052
return batchspec_t(new_batch_type, els_left, size_left, end_time);
5153
}
5254

55+
batchspec_t batchspec_t::with_at_most(uint64_t _max_els) const {
56+
int64_t max_els = std::min(uint64_t(std::numeric_limits<int64_t>::max()), _max_els);
57+
return batchspec_t(
58+
batch_type,
59+
std::max<int64_t>(1, std::min(els_left, max_els)),
60+
size_left,
61+
end_time);
62+
}
63+
5364
batcher_t batchspec_t::to_batcher() const {
5465
microtime_t real_end_time =
5566
batch_type == batch_type_t::NORMAL && end_time > current_microtime()

‎src/rdb_protocol/batching.hpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ class batchspec_t {
5858
static batchspec_t empty() { return batchspec_t(); }
5959
batch_type_t get_batch_type() const { return batch_type; }
6060
batchspec_t with_new_batch_type(batch_type_t new_batch_type) const;
61+
batchspec_t with_at_most(uint64_t max_els) const;
6162
batcher_t to_batcher() const;
6263
RDB_MAKE_ME_SERIALIZABLE_4(batch_type, els_left, size_left, end_time);
6364
private:

‎src/rdb_protocol/datum.cc‎

Lines changed: 49 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,7 @@ void datum_t::init_json(cJSON *json) {
140140
case cJSON_String: {
141141
init_str();
142142
*r_str = json->valuestring;
143+
check_str_validity(*r_str);
143144
} break;
144145
case cJSON_Array: {
145146
init_array();
@@ -935,6 +936,10 @@ void datum_t::init_from_pb(const Datum *d) {
935936
*r_str = d->r_str();
936937
check_str_validity(*r_str);
937938
} break;
939+
case Datum::R_JSON: {
940+
scoped_cJSON_t cjson(cJSON_Parse(d->r_str().c_str()));
941+
init_json(cjson.get());
942+
} break;
938943
case Datum::R_ARRAY: {
939944
init_array();
940945
for (int i = 0; i < d->r_array_size(); ++i) {
@@ -979,43 +984,51 @@ bool datum_t::key_is_truncated(const store_key_t &key) {
979984
}
980985
}
981986

982-
void datum_t::write_to_protobuf(Datum *d) const {
983-
switch (get_type()) {
984-
case R_NULL: {
985-
d->set_type(Datum::R_NULL);
986-
} break;
987-
case R_BOOL: {
988-
d->set_type(Datum::R_BOOL);
989-
d->set_r_bool(r_bool);
990-
} break;
991-
case R_NUM: {
992-
d->set_type(Datum::R_NUM);
993-
// so we can use `isfinite` in a GCC 4.4.3-compatible way
994-
using namespace std; // NOLINT(build/namespaces)
995-
r_sanity_check(isfinite(r_num));
996-
d->set_r_num(r_num);
997-
} break;
998-
case R_STR: {
999-
d->set_type(Datum::R_STR);
1000-
d->set_r_str(*r_str);
1001-
} break;
1002-
case R_ARRAY: {
1003-
d->set_type(Datum::R_ARRAY);
1004-
for (size_t i = 0; i < r_array->size(); ++i) {
1005-
(*r_array)[i]->write_to_protobuf(d->add_r_array());
987+
void datum_t::write_to_protobuf(Datum *d, use_json_t use_json) const {
988+
switch (use_json) {
989+
case use_json_t::NO: {
990+
switch (get_type()) {
991+
case R_NULL: {
992+
d->set_type(Datum::R_NULL);
993+
} break;
994+
case R_BOOL: {
995+
d->set_type(Datum::R_BOOL);
996+
d->set_r_bool(r_bool);
997+
} break;
998+
case R_NUM: {
999+
d->set_type(Datum::R_NUM);
1000+
// so we can use `isfinite` in a GCC 4.4.3-compatible way
1001+
using namespace std; // NOLINT(build/namespaces)
1002+
r_sanity_check(isfinite(r_num));
1003+
d->set_r_num(r_num);
1004+
} break;
1005+
case R_STR: {
1006+
d->set_type(Datum::R_STR);
1007+
d->set_r_str(*r_str);
1008+
} break;
1009+
case R_ARRAY: {
1010+
d->set_type(Datum::R_ARRAY);
1011+
for (size_t i = 0; i < r_array->size(); ++i) {
1012+
(*r_array)[i]->write_to_protobuf(d->add_r_array(), use_json);
1013+
}
1014+
} break;
1015+
case R_OBJECT: {
1016+
d->set_type(Datum::R_OBJECT);
1017+
// We use rbegin and rend so that things print the way we expect.
1018+
for (auto it = r_object->rbegin(); it != r_object->rend(); ++it) {
1019+
Datum_AssocPair *ap = d->add_r_object();
1020+
ap->set_key(it->first);
1021+
it->second->write_to_protobuf(ap->mutable_val(), use_json);
1022+
}
1023+
} break;
1024+
case UNINITIALIZED: // fallthru
1025+
default: unreachable();
10061026
}
10071027
} break;
1008-
case R_OBJECT: {
1009-
d->set_type(Datum::R_OBJECT);
1010-
// We use rbegin and rend so that things print the way we expect.
1011-
for (std::map<std::string, counted_t<const datum_t> >::const_reverse_iterator
1012-
it = r_object->rbegin(); it != r_object->rend(); ++it) {
1013-
Datum_AssocPair *ap = d->add_r_object();
1014-
ap->set_key(it->first);
1015-
it->second->write_to_protobuf(ap->mutable_val());
1016-
}
1028+
case use_json_t::YES: {
1029+
d->set_type(Datum::R_JSON);
1030+
d->set_r_str(as_json().PrintUnformatted());
10171031
} break;
1018-
case UNINITIALIZED: // fallthru
10191032
default: unreachable();
10201033
}
10211034
}
@@ -1283,8 +1296,8 @@ void wire_datum_map_t::finalize() {
12831296
r_sanity_check(state == COMPILED);
12841297
while (!map.empty()) {
12851298
map_pb.push_back(std::make_pair(Datum(), Datum()));
1286-
map.begin()->first->write_to_protobuf(&map_pb.back().first);
1287-
map.begin()->second->write_to_protobuf(&map_pb.back().second);
1299+
map.begin()->first->write_to_protobuf(&map_pb.back().first, use_json_t::NO);
1300+
map.begin()->second->write_to_protobuf(&map_pb.back().second, use_json_t::NO);
12881301
map.erase(map.begin());
12891302
}
12901303
state = SERIALIZABLE;

‎src/rdb_protocol/datum.hpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,8 @@ enum throw_bool_t { NOTHROW = 0, THROW = 1};
4646
// CLOBBER: Overwrite existing values.
4747
enum clobber_bool_t { NOCLOBBER = 0, CLOBBER = 1};
4848

49+
enum class use_json_t { NO = 0, YES = 1 };
50+
4951
// A `datum_t` is basically a JSON value, although we may extend it later.
5052
class datum_t : public slow_atomic_countable_t<datum_t> {
5153
public:
@@ -78,7 +80,7 @@ class datum_t : public slow_atomic_countable_t<datum_t> {
7880

7981
~datum_t();
8082

81-
void write_to_protobuf(Datum *out) const;
83+
void write_to_protobuf(Datum *out, use_json_t use_json) const;
8284

8385
type_t get_type() const;
8486
bool is_ptype() const;

‎src/rdb_protocol/datum_stream.cc‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -865,17 +865,18 @@ concatmap_datum_stream_t::next_batch_impl(env_t *env, const batchspec_t &batchsp
865865
}
866866

867867
// SLICE_DATUM_STREAM_T
868-
slice_datum_stream_t::slice_datum_stream_t(size_t _left, size_t _right,
868+
slice_datum_stream_t::slice_datum_stream_t(uint64_t _left, uint64_t _right,
869869
counted_t<datum_stream_t> _src)
870870
: wrapper_datum_stream_t(_src), index(0),
871871
left(_left), right(_right) { }
872872

873873
std::vector<counted_t<const datum_t> >
874-
slice_datum_stream_t::next_batch_impl(env_t *env, const batchspec_t &batchspec) {
874+
slice_datum_stream_t::next_batch_impl(env_t *env, const batchspec_t &_batchspec) {
875875
if (left >= right || index >= right) {
876876
return std::vector<counted_t<const datum_t> >();
877877
}
878878

879+
const batchspec_t batchspec = _batchspec.with_at_most(right - index);
879880
while (index < left) {
880881
std::vector<counted_t<const datum_t> > v = source->next_batch(env, batchspec);
881882
if (v.size() == 0) {
@@ -897,7 +898,7 @@ slice_datum_stream_t::next_batch_impl(env_t *env, const batchspec_t &batchspec)
897898
while (index < right) {
898899
std::vector<counted_t<const datum_t> > v = source->next_batch(env, batchspec);
899900
if (v.size() == 0) {
900-
return v;
901+
break;
901902
}
902903
index += v.size();
903904
if (index > right) {
@@ -909,6 +910,10 @@ slice_datum_stream_t::next_batch_impl(env_t *env, const batchspec_t &batchspec)
909910
return std::vector<counted_t<const datum_t> >();
910911
}
911912

913+
bool slice_datum_stream_t::is_exhausted() const {
914+
return left >= right || index >= right || source->is_exhausted();
915+
}
916+
912917
// ZIP_DATUM_STREAM_T
913918
zip_datum_stream_t::zip_datum_stream_t(counted_t<datum_stream_t> _src)
914919
: wrapper_datum_stream_t(_src) { }

0 commit comments

Comments
 (0)