Timezone conversion ambiguity [JIRA: CLIENTS-897]

[korry8911]

I am writing the following to Riak TS:

['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 12, 0, 0), 0]

When reading this item from Riak TS I get back the following:

['field1_val', 'field2_val', 1420142400000, 0]

Converting 1420142400000 to datetime I get:

datetime.datetime.utcfromtimestamp(1420142400)
datetime.datetime(2015, 1, 1, 20, 0)

Now if I try to see the unix epoch time in milliseconds for datetime.datetime(2015, 1, 1, 12, 0, 0), I get:

1420113600000

I am using this function to convert to unix epoch milliseconds:

def unix_time_millis(dt):
    td = dt - datetime.datetime.utcfromtimestamp(0)
    return int(td.total_seconds() * 1000.0)

Somewhere there is a timezone issue. Not sure where.

For completeness, this is my actually code snippet:

test_row_in = ['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 12, 0, 0), 0]
    test_rdd = spark_context.parallelize([test_row_in])
    print(test_rdd.collect())
    test_df = test_rdd.toDF(['field1', 'field2', 'datetime', 'data'])

    one_second = datetime.timedelta(seconds=1)
    one_day = datetime.timedelta(days=1)
    start = datetime.datetime(2015, 1, 1, 12, 0, 0) - one_day
    end = datetime.datetime(2015, 1, 1, 12, 0, 0) + one_day

    fmt =   """
                select * from {table_name}
                where datetime >= {start_date}
                AND datetime <=  {end_date}
                AND field1 = '{field1}'
                AND field2 = '{field2}'          
            """
    query = fmt.format(table_name=riak_ts_table_name, start_date=unix_time_millis(start), end_date=unix_time_millis(end), field1='field1_val', field2='field2_val')


    assert retry_func_with_timeout(func=test_df.write.format('org.apache.spark.sql.riak').mode('Append').save,
                                   times=10, 
                                   timeout=30, 
                                   signal=True, 
                                   args=[riak_ts_table_name], 
                                   use_condition=False, 
                                   condition_func=None, 
                                   condition_val=None,
                                   test_func=None,
                                   test_args=None
                                   )[0] == True



    assert retry_func_with_timeout(func=riak_ts_table.query, 
                                   times=10, 
                                   timeout=3, 
                                   signal=True, 
                                   args=[query], 
                                   use_condition=True, 
                                   condition_func=ts_query_condition, 
                                   condition_val=test_rdd.collect(),
                                   test_func=None,
                                   test_args=None
                                   )[0] == True

I am running riak (2.5.3), on OSX El Capitan 10.11.3 (15D21)

[lukebakken]

• "when reading this item" - how are you reading that particular row back?
• Try using the function in riak.util instead of your own:

from riak.util import unix_time_millis

[lukebakken]

For what it's worth, I added tests in b5f0bdb and can't reproduce.

[korry8911]

I am reading the row back with this statement:

assert retry_func_with_timeout(func=riak_ts_table.query, 
                                   times=10, 
                                   timeout=3, 
                                   signal=True, 
                                   args=[query], 
                                   use_condition=True, 
                                   condition_func=ts_query_condition, 
                                   condition_val=test_rdd.collect(),
                                   test_func=None,
                                   test_args=None
                                   )[0] == True

Its running a riak_ts_table.query(query) where query is:

fmt =   """
                select * from {table_name}
                where datetime >= {start_date}
                AND datetime <=  {end_date}
                AND field1 = '{field1}'
                AND field2 = '{field2}'          
            """
    query = fmt.format(table_name=riak_ts_table_name, start_date=unix_time_millis(start), end_date=unix_time_millis(end), field1='field1_val', field2='field2_val')

[korry8911]

Here is a notebook of the issue. Note that result = client.ts_get(riak_ts_table_name, ['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 12, 0, 0)]) return nothing, even though this ['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 12, 0, 0), 0] is the row I am writing in.

from riak.util import unix_time_millis
import datetime
import riak
import time
import findspark
findspark.init()
import pyspark
import pyspark_riak
import os
from pyspark import SparkContext, SparkConf, SQLContext, Row
os.environ['SPARK_CLASSPATH'] = '/Users/basho/.m2/repository/com/basho/riak/spark-riak-connector/1.5.1-SNAPSHOT/spark-riak-connector-1.5.1-SNAPSHOT-uber.jar'

client = riak.RiakClient(host='172.17.0.2', pb_port='8087')

riak_ts_table_name = 'spark-riak-%d' % int(time.time())
riak_ts_table = client.table(riak_ts_table_name)

create_sql = """CREATE TABLE %(table_name)s (
    field1 varchar not null,
    field2 varchar not null,
    datetime timestamp not null,
    data sint64,
    PRIMARY KEY ((field1, field2, quantum(datetime, 24, h)), field1, field2, datetime))
    """ % ({'table_name': riak_ts_table_name})
riak_ts_table.query(create_sql)

<riak.ts_object.TsObject at 0x10f6df950>

sc.stop()
conf = (SparkConf().setMaster("local[*]").setAppName("pytest-pyspark-local-testing"))
conf.set('spark.riak.connection.host', '172.17.0.2:8087')
sc = SparkContext(conf=conf)
sc.setLogLevel("ERROR")
pyspark_riak.riak_context(sc)
sqlContext = SQLContext(sc)

test_row_in = ['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 12, 0, 0), 0]
test_rdd = sc.parallelize([test_row_in])
test_df = test_rdd.toDF(['field1', 'field2', 'datetime', 'data'])

test_df.write.format('org.apache.spark.sql.riak').mode('Append').save(riak_ts_table_name)

one_second = datetime.timedelta(seconds=1)
one_day = datetime.timedelta(days=1)
start = datetime.datetime(2015, 1, 1, 12, 0, 0) - one_day
end = datetime.datetime(2015, 1, 1, 12, 0, 0) + one_day

fmt =   """
                select * from {table_name}
                where datetime >= {start_date}
                AND datetime <=  {end_date}
                AND field1 = '{field1}'
                AND field2 = '{field2}'          
            """
query = fmt.format(table_name=riak_ts_table_name, start_date=unix_time_millis(start), end_date=unix_time_millis(end), field1='field1_val', field2='field2_val')

result = riak_ts_table.query(query)

result.rows

[['field1_val', 'field2_val', 1420142400000, 0]]

datetime.datetime.fromtimestamp(result.rows[0][2]/1000.0)

datetime.datetime(2015, 1, 1, 12, 0)

datetime.datetime.utcfromtimestamp(result.rows[0][2]/1000.0)

datetime.datetime(2015, 1, 1, 20, 0)

result = client.ts_get(riak_ts_table_name, ['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 12, 0, 0)])

result.rows

[]

result = client.ts_get(riak_ts_table_name, ['field1_val', 'field2_val', datetime.datetime(2015, 1, 1, 20, 0, 0)])

result.rows

[['field1_val', 'field2_val', 1420142400000, 0]]

[korry8911]

So it looks like when writing a datetime in to Riak TS, it takes the datetime object as being in my local timezone and converts it to GMT. But when I query the table, I have to query in GMT. The query is not converting my datetime query value to GMT but is comparing it to GMT. Maybe not a bug but still confusing, especially since there is no clear documentation on this. If Riak TS converts my data to GMT at write time, it seems like it should convert my querys to GMT as well.

[lukebakken]

There's a lot of code in this issue, so let's narrow this down a bit.

• When you store a datetime in Riak TS, the Python client will convert that value to UTC milliseconds since the epoch and store that integer in Riak TS, which is what you're seeing.
• When you use ts_get to fetch a row of data, and you use a datetime as part of the key, the Python client will convert that datetime to UTC milliseconds (you can see the test here).
• When you use ts_query to issue an SQL query, all the datetime objects you use as part of the query must be converted to UTC milliseconds using the riak.util.unix_time_millis method. Riak 1.4 should support passing datetime values as ISO8601 strings to be converted server-side, but for now, you must convert the datetime objects to integers yourself.

So, in your case, can we get a simple set of ts_store and a ts_get statement that reproduces your issue?

[korry8911]

So after further investigation, the client seems to work as you have indicated it should. My main concern is that there are a couple of different ways to query RiakTS using the Riak python client. The documentation for these two ways (ts_get and ts_query) and the details for each way of querying are poor. It might be a good idea to have somewhere that is easily findable (not buried deep in some api docs) that has examples of ts_get and ts_query and the different ways they should be used. Perhaps I was looking in the wrong places, but I could not find anything other than the api docs here https://basho.github.io/riak-python-client/index.html and the examples here http://docs.basho.com/riak/ts/1.3.0/using/querying/. In these examples, it is implicitly assumed that the datetime objects are already in the GMT +0 timezone. This is problematic without notice to the user. So basically, the problem boils down to docs and the lack of good docs. @mdigan @alexmoore @ph07 @DSomogyi

[lukebakken]

@korry8911 - ts_get should not require a datetime in UTC, as you can see in the test I linked to. We should investigate that further in your environment.

As for the docs, yes they could be improved.

[lukebakken]

In summary: unix_time_millis expects a datetime that represents a UTC value. If the user passes in a datetime created by datetime.now(), for instance, it will be the local time, but without any time zone info, because Python date/time handling sucks.

One idea for a fix is to see if dt has a tzinfo attached, and use that.