From ec2fb60df664092f7907fc308c0b8b6caabcb8a0 Mon Sep 17 00:00:00 2001
From: Vikas Kedigehalli <vikasrk@google.com>
Date: Wed, 7 Jun 2017 16:26:21 -0700
Subject: [PATCH] Make BytesCoder to be a known type

---
 sdks/python/apache_beam/coders/coders.py                  | 5 +++++
 sdks/python/apache_beam/runners/worker/operation_specs.py | 4 ++++
 2 files changed, 9 insertions(+)

diff --git a/sdks/python/apache_beam/coders/coders.py b/sdks/python/apache_beam/coders/coders.py
index f40045d142ff..f3e0b432e51c 100644
--- a/sdks/python/apache_beam/coders/coders.py
+++ b/sdks/python/apache_beam/coders/coders.py
@@ -286,6 +286,11 @@ def _create_impl(self):
   def is_deterministic(self):
     return True
 
+  def as_cloud_object(self):
+    return {
+        '@type': 'kind:bytes',
+    }
+
   def __eq__(self, other):
     return type(self) == type(other)
 
diff --git a/sdks/python/apache_beam/runners/worker/operation_specs.py b/sdks/python/apache_beam/runners/worker/operation_specs.py
index db5eb765598b..b8d19a142756 100644
--- a/sdks/python/apache_beam/runners/worker/operation_specs.py
+++ b/sdks/python/apache_beam/runners/worker/operation_specs.py
@@ -339,6 +339,10 @@ def get_coder_from_spec(coder_spec):
     assert len(coder_spec['component_encodings']) == 1
     return coders.coders.LengthPrefixCoder(
         get_coder_from_spec(coder_spec['component_encodings'][0]))
+  elif coder_spec['@type'] == 'kind:bytes':
+    assert ('component_encodings' not in coder_spec
+            or len(coder_spec['component_encodings'] == 0))
+    return coders.BytesCoder()
 
   # We pass coders in the form "<coder_name>$<pickled_data>" to make the job
   # description JSON more readable.