Skip to content

Commit a12f6fd

Browse files
lukebakkenlukebakken
committed
Merge pull request #455 from basho/fixes/lrb/timestamp-encoding-gh-454
READY: Make Riak TS timestamp conversion optional
2 parents 14e2294 + 7b4b01a commit a12f6fd

15 files changed

Lines changed: 186 additions & 235 deletions

‎riak/codecs/pbuf.py‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -773,7 +773,8 @@ def encode_timeseries_query(self, table, query, interpolations=None):
773773
rc = riak.pb.messages.MSG_CODE_TS_QUERY_RESP
774774
return Msg(mc, req.SerializeToString(), rc)
775775

776-
def decode_timeseries(self, resp, tsobj):
776+
def decode_timeseries(self, resp, tsobj,
777+
convert_timestamp=False):
777778
"""
778779
Fills an TsObject with the appropriate data and
779780
metadata from a TsGetResp / TsQueryResp.
@@ -783,6 +784,8 @@ def decode_timeseries(self, resp, tsobj):
783784
riak.pb.riak_ts_pb2.TsGetResp
784785
:param tsobj: a TsObject
785786
:type tsobj: TsObject
787+
:param convert_timestamp: Convert timestamps to datetime objects
788+
:type tsobj: boolean
786789
"""
787790
if resp.columns is not None:
788791
col_names = []
@@ -798,7 +801,7 @@ def decode_timeseries(self, resp, tsobj):
798801
for row in resp.rows:
799802
tsobj.rows.append(
800803
self.decode_timeseries_row(
801-
row, resp.columns))
804+
row, resp.columns, convert_timestamp))
802805

803806
def decode_timeseries_col_type(self, col_type):
804807
# NB: these match the atom names for column types
@@ -816,7 +819,8 @@ def decode_timeseries_col_type(self, col_type):
816819
msg = 'could not decode column type: {}'.format(col_type)
817820
raise RiakError(msg)
818821

819-
def decode_timeseries_row(self, tsrow, tscols=None):
822+
def decode_timeseries_row(self, tsrow, tscols=None,
823+
convert_timestamp=False):
820824
"""
821825
Decodes a TsRow into a list
822826
@@ -850,8 +854,10 @@ def decode_timeseries_row(self, tsrow, tscols=None):
850854
if col and col.type != TsColumnType.Value('TIMESTAMP'):
851855
raise TypeError('expected TIMESTAMP column')
852856
else:
853-
dt = datetime_from_unix_time_millis(
854-
cell.timestamp_value)
857+
dt = cell.timestamp_value
858+
if convert_timestamp:
859+
dt = datetime_from_unix_time_millis(
860+
cell.timestamp_value)
855861
row.append(dt)
856862
elif cell.HasField('boolean_value'):
857863
if col and col.type != TsColumnType.Value('BOOLEAN'):

‎riak/codecs/ttb.py‎

Lines changed: 21 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
from riak import RiakError
88
from riak.codecs import Codec, Msg
9+
from riak.pb.messages import MSG_CODE_TS_TTB_MSG
910
from riak.ts_object import TsColumns
1011
from riak.util import bytes_to_str, unix_time_millis, \
1112
datetime_from_unix_time_millis
@@ -23,9 +24,6 @@
2324
tsdelreq_a = Atom('tsdelreq')
2425
timestamp_a = Atom('timestamp')
2526

26-
# TODO RTS-842
27-
MSG_CODE_TS_TTB = 104
28-
2927

3028
class TtbCodec(Codec):
3129
'''
@@ -36,7 +34,7 @@ def __init__(self, **unused_args):
3634
super(TtbCodec, self).__init__(**unused_args)
3735

3836
def parse_msg(self, msg_code, data):
39-
if msg_code != MSG_CODE_TS_TTB:
37+
if msg_code != MSG_CODE_TS_TTB_MSG:
4038
raise RiakError("TTB can't parse code: {}".format(msg_code))
4139
if len(data) > 0:
4240
decoded = decode(data)
@@ -61,6 +59,7 @@ def encode_to_ts_cell(self, cell):
6159
else:
6260
if isinstance(cell, datetime.datetime):
6361
ts = unix_time_millis(cell)
62+
# logging.debug('encoded datetime %s as %s', cell, ts)
6463
return ts
6564
elif isinstance(cell, bool):
6665
return cell
@@ -84,19 +83,19 @@ def encode_timeseries_keyreq(self, table, key, is_delete=False):
8483
else:
8584
raise ValueError("key must be a list")
8685

87-
mc = MSG_CODE_TS_TTB
88-
rc = MSG_CODE_TS_TTB
86+
mc = MSG_CODE_TS_TTB_MSG
87+
rc = MSG_CODE_TS_TTB_MSG
8988
req_atom = tsgetreq_a
9089
if is_delete:
9190
req_atom = tsdelreq_a
9291

93-
# TODO RTS-842 timeout is last
92+
# TODO FUTURE add timeout as last param
9493
req = req_atom, table.name, \
9594
[self.encode_to_ts_cell(k) for k in key_vals], udef_a
9695
return Msg(mc, encode(req), rc)
9796

9897
def validate_timeseries_put_resp(self, resp_code, resp):
99-
if resp is None and resp_code == MSG_CODE_TS_TTB:
98+
if resp is None and resp_code == MSG_CODE_TS_TTB_MSG:
10099
return True
101100
if resp is not None:
102101
return True
@@ -123,8 +122,8 @@ def encode_timeseries_put(self, tsobj):
123122
req_r.append(self.encode_to_ts_cell(cell))
124123
req_rows.append(tuple(req_r))
125124
req = tsputreq_a, tsobj.table.name, [], req_rows
126-
mc = MSG_CODE_TS_TTB
127-
rc = MSG_CODE_TS_TTB
125+
mc = MSG_CODE_TS_TTB_MSG
126+
rc = MSG_CODE_TS_TTB_MSG
128127
return Msg(mc, encode(req), rc)
129128
else:
130129
raise RiakError("TsObject requires a list of rows")
@@ -135,11 +134,12 @@ def encode_timeseries_query(self, table, query, interpolations=None):
135134
q = q.format(table=table.name)
136135
tsi = tsinterpolation_a, q, []
137136
req = tsqueryreq_a, tsi, False, []
138-
mc = MSG_CODE_TS_TTB
139-
rc = MSG_CODE_TS_TTB
137+
mc = MSG_CODE_TS_TTB_MSG
138+
rc = MSG_CODE_TS_TTB_MSG
140139
return Msg(mc, encode(req), rc)
141140

142-
def decode_timeseries(self, resp_ttb, tsobj):
141+
def decode_timeseries(self, resp_ttb, tsobj,
142+
convert_timestamp=False):
143143
"""
144144
Fills an TsObject with the appropriate data and
145145
metadata from a TTB-encoded TsGetResp / TsQueryResp.
@@ -148,6 +148,8 @@ def decode_timeseries(self, resp_ttb, tsobj):
148148
:type resp_ttb: TTB-encoded tsqueryrsp or tsgetresp
149149
:param tsobj: a TsObject
150150
:type tsobj: TsObject
151+
:param convert_timestamp: Convert timestamps to datetime objects
152+
:type tsobj: boolean
151153
"""
152154
if resp_ttb is None:
153155
return tsobj
@@ -170,7 +172,8 @@ def decode_timeseries(self, resp_ttb, tsobj):
170172
tsobj.rows = []
171173
for resp_row in resp_rows:
172174
tsobj.rows.append(
173-
self.decode_timeseries_row(resp_row, resp_coltypes))
175+
self.decode_timeseries_row(resp_row, resp_coltypes,
176+
convert_timestamp))
174177
else:
175178
raise RiakError(
176179
"Expected 3-tuple in response, got: {}".format(resp_data))
@@ -182,14 +185,16 @@ def decode_timeseries_cols(self, cnames, ctypes):
182185
ctypes = [str(ctype) for ctype in ctypes]
183186
return TsColumns(cnames, ctypes)
184187

185-
def decode_timeseries_row(self, tsrow, tsct):
188+
def decode_timeseries_row(self, tsrow, tsct, convert_timestamp=False):
186189
"""
187190
Decodes a TTB-encoded TsRow into a list
188191
189192
:param tsrow: the TTB decoded TsRow to decode.
190193
:type tsrow: TTB dncoded row
191194
:param tsct: the TTB decoded column types (atoms).
192195
:type tsct: list
196+
:param convert_timestamp: Convert timestamps to datetime objects
197+
:type tsobj: boolean
193198
:rtype list
194199
"""
195200
row = []
@@ -199,7 +204,7 @@ def decode_timeseries_row(self, tsrow, tsct):
199204
elif isinstance(cell, list) and len(cell) == 0:
200205
row.append(None)
201206
else:
202-
if tsct[i] == timestamp_a:
207+
if convert_timestamp and tsct[i] == timestamp_a:
203208
row.append(datetime_from_unix_time_millis(cell))
204209
else:
205210
row.append(cell)

‎riak/pb/messages.py‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -81,9 +81,7 @@
8181
MSG_CODE_TS_COVERAGE_RESP = 101
8282
MSG_CODE_TS_COVERAGE_ENTRY = 102
8383
MSG_CODE_TS_RANGE = 103
84-
MSG_CODE_TS_TTB_PUT_REQ = 104
85-
MSG_CODE_TOGGLE_ENCODING_REQ = 110
86-
MSG_CODE_TOGGLE_ENCODING_RESP = 111
84+
MSG_CODE_TS_TTB_MSG = 104
8785
MSG_CODE_AUTH_REQ = 253
8886
MSG_CODE_AUTH_RESP = 254
8987
MSG_CODE_START_TLS = 255
@@ -168,9 +166,7 @@
168166
MSG_CODE_TS_COVERAGE_RESP: riak.pb.riak_ts_pb2.TsCoverageResp,
169167
MSG_CODE_TS_COVERAGE_ENTRY: riak.pb.riak_ts_pb2.TsCoverageEntry,
170168
MSG_CODE_TS_RANGE: riak.pb.riak_ts_pb2.TsRange,
171-
MSG_CODE_TS_TTB_PUT_REQ: riak.pb.riak_ts_pb2.TsTtbPutReq,
172-
MSG_CODE_TOGGLE_ENCODING_REQ: riak.pb.riak_pb2.RpbToggleEncodingReq,
173-
MSG_CODE_TOGGLE_ENCODING_RESP: riak.pb.riak_pb2.RpbToggleEncodingResp,
169+
MSG_CODE_TS_TTB_MSG: None,
174170
MSG_CODE_AUTH_REQ: riak.pb.riak_pb2.RpbAuthReq,
175171
MSG_CODE_AUTH_RESP: None,
176172
MSG_CODE_START_TLS: None

‎riak/pb/riak_pb2.py‎

Lines changed: 1 addition & 71 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)