beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From al...@apache.org
Subject [1/2] beam git commit: Fix WindowValueCoder for large timestamps
Date Tue, 13 Jun 2017 17:11:42 GMT
Repository: beam
Updated Branches:
  refs/heads/master e906fe9c3 -> c33e9b446


Fix WindowValueCoder for large timestamps


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/ee728f1b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/ee728f1b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/ee728f1b

Branch: refs/heads/master
Commit: ee728f1b2f617dac8e5cd729cacf1a46911021e0
Parents: e906fe9
Author: Vikas Kedigehalli <vikasrk@google.com>
Authored: Mon Jun 12 23:11:22 2017 -0700
Committer: Ahmet Altay <altay@google.com>
Committed: Tue Jun 13 10:11:20 2017 -0700

----------------------------------------------------------------------
 sdks/python/apache_beam/coders/coder_impl.py         | 4 ++++
 sdks/python/apache_beam/coders/coders_test_common.py | 8 ++++++++
 2 files changed, 12 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/ee728f1b/sdks/python/apache_beam/coders/coder_impl.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coder_impl.py b/sdks/python/apache_beam/coders/coder_impl.py
index 10298bf..2670250 100644
--- a/sdks/python/apache_beam/coders/coder_impl.py
+++ b/sdks/python/apache_beam/coders/coder_impl.py
@@ -710,6 +710,10 @@ class WindowedValueCoderImpl(StreamCoderImpl):
       timestamp = MAX_TIMESTAMP.micros
     else:
       timestamp *= 1000
+      if timestamp > MAX_TIMESTAMP.micros:
+        timestamp = MAX_TIMESTAMP.micros
+      if timestamp < MIN_TIMESTAMP.micros:
+        timestamp = MIN_TIMESTAMP.micros
 
     windows = self._windows_coder.decode_from_stream(in_stream, True)
     # Read PaneInfo encoded byte.

http://git-wip-us.apache.org/repos/asf/beam/blob/ee728f1b/sdks/python/apache_beam/coders/coders_test_common.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coders_test_common.py b/sdks/python/apache_beam/coders/coders_test_common.py
index c9b67b3..577c53a 100644
--- a/sdks/python/apache_beam/coders/coders_test_common.py
+++ b/sdks/python/apache_beam/coders/coders_test_common.py
@@ -23,6 +23,8 @@ import unittest
 
 import dill
 
+from apache_beam.transforms.window import GlobalWindow
+from apache_beam.utils.timestamp import MIN_TIMESTAMP
 import observable
 from apache_beam.transforms import window
 from apache_beam.utils import timestamp
@@ -287,6 +289,12 @@ class CodersTest(unittest.TestCase):
     # Test binary representation
     self.assertEqual('\x7f\xdf;dZ\x1c\xac\t\x00\x00\x00\x01\x0f\x01',
                      coder.encode(window.GlobalWindows.windowed_value(1)))
+
+    # Test decoding large timestamp
+    self.assertEqual(
+        coder.decode('\x7f\xdf;dZ\x1c\xac\x08\x00\x00\x00\x01\x0f\x00'),
+        windowed_value.create(0, MIN_TIMESTAMP.micros, (GlobalWindow(),)))
+
     # Test unnested
     self.check_coder(
         coders.WindowedValueCoder(coders.VarIntCoder()),


Mime
View raw message