Add TimestampCoder for timeutil.Timestamp objects

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=118044798

Author: charlesccychen

google/cloud/dataflow/coders/coders.py

@@ -15,6 +15,7 @@
 """Collection of useful coders."""
 
 import base64
+import collections
 import cPickle as pickle
 import struct
 
@@ -27,10 +28,12 @@
   # Import dill from the pickler module to make sure our monkey-patching of dill
   # occurs.
   from google.cloud.dataflow.internal.pickler import dill
+  from google.cloud.dataflow.transforms.timeutil import Timestamp
 except ImportError:
   # We fall back to using the stock dill library in tests that don't use the
   # full Python SDK.
   import dill
+  Timestamp = collections.namedtuple('Timestamp', 'micros')
 
 
 def serialize_coder(coder):
@@ -236,6 +239,20 @@ def is_deterministic(self):
     return True
 
 
+# TODO(ccy): Write a Cython implementation of TimestampCoder.
+class TimestampCoder(Coder):
+  """A coder used for timeutil.Timestamp values."""
+
+  def encode(self, value):
+    return struct.pack('<q', value.micros)
+
+  def decode(self, encoded):
+    return Timestamp(micros=struct.unpack('<q', encoded)[0])
+
+  def is_deterministic(self):
+    return True
+
+
 def maybe_dill_dumps(o):
   """Pickle using cPickle or the Dill pickler as a fallback."""
   # We need to use the dill pickler for objects of certain custom classes,
@@ -428,7 +445,7 @@ class WindowedValueCoder(FastCoder):
   def __init__(self, wrapped_value_coder, timestamp_coder=None,
                window_coder=None):
     if not timestamp_coder:
-      timestamp_coder = FloatCoder()
+      timestamp_coder = TimestampCoder()
     if not window_coder:
       window_coder = PickleCoder()
     self.wrapped_value_coder = wrapped_value_coder

google/cloud/dataflow/coders/coders_test_common.py

@@ -50,9 +50,14 @@ def tearDownClass(cls):
                    for c in coders.__dict__.values()
                    if isinstance(c, type) and issubclass(c, coders.Coder) and
                    'Base' not in c.__name__)
-    standard -= set([coders.Coder, coders.ToStringCoder, coders.FloatCoder,
-                     coders.Base64PickleCoder, coders.FastCoder,
-                     coders.WindowCoder, coders.WindowedValueCoder])
+    standard -= set([coders.Coder,
+                     coders.FastCoder,
+                     coders.Base64PickleCoder,
+                     coders.FloatCoder,
+                     coders.TimestampCoder,
+                     coders.ToStringCoder,
+                     coders.WindowCoder,
+                     coders.WindowedValueCoder])
     assert not standard - cls.seen, standard - cls.seen
     assert not standard - cls.seen_nested, standard - cls.seen_nested
 
@@ -124,6 +129,16 @@ def test_float_coder(self):
                      *[float(2 ** (0.1 * x)) for x in range(-100, 100)])
     self.check_coder(coders.FloatCoder(), float('-Inf'), float('Inf'))
 
+  def test_timestamp_coder(self):
+    self.check_coder(coders.TimestampCoder(),
+                     *[coders.Timestamp(micros=x) for x in range(-100, 100)])
+    self.check_coder(coders.TimestampCoder(),
+                     coders.Timestamp(micros=-1234567890),
+                     coders.Timestamp(micros=1234567890))
+    self.check_coder(coders.TimestampCoder(),
+                     coders.Timestamp(micros=-1234567890123456789),
+                     coders.Timestamp(micros=1234567890123456789))
+
   def test_tuple_coder(self):
     self.check_coder(
         coders.TupleCoder((coders.VarIntCoder(), coders.BytesCoder())),