Repository: tensorflow/datasets (python)

Issue:
FeatureConnector serialization broken for byte strings
**Short description**
I am currently writing a custom FeatureConnector for RLE encoded segmentations. In order to save some storage space I am going to use the byte encoding of the numpy arrays and `tf.io.decode_raw` in order to decode the information (since tf.train.Example only supports int64 for integer data types). However, this line breaks the serialization of the byte string: 
https://github.com/tensorflow/datasets/blob/c8c00f0843166775b44d3eba2e2171b44d2cb689/tensorflow_datasets/core/file_format_adapter.py#L397
One could argue, that this is a Tensorflow compat module bug, however, bytes returned from a FeatureConnector should directly be passed into the Protobuf Message instead of doing unnecessary conversions. 

**Environment information**
* Operating System: Ubuntu 18.04
* Python version: 3.6
* `tensorflow-datasets`/`tfds-nightly` version: tensorflow-datasets 1.0.1
* `tensorflow`/`tensorflow-gpu`/`tf-nightly`/`tf-nightly-gpu` version: tensorflow-gpu 1.13.1

**Reproduction instructions**

```python
import numpy as np
import tensorflow as tf
tf.enable_eager_execution()
arr = np.zeros((64, 64), dtype=np.uint8)
bytes = arr.tobytes()
print(len(bytes))  # 4096
v = np.asarray(bytes).flatten()  # tensorflow_datasets/core/file_format_adapter.py#390
bytes = tf.compat.as_bytes(v[0])  # tensorflow_datasets/core/file_format_adapter.py#397
print(len(bytes)) # 0
```

Pull request description:
Fix #237: byte strings got truncated by tf.compat.as_bytes
Fix for issue #237, where byte strings from a custom feature connector get truncated by `tf.compat.as_bytes`.

Patch:
diff --git a/tensorflow_datasets/core/file_format_adapter.py b/tensorflow_datasets/core/file_format_adapter.py
index 824ac73f2..476c0c488 100644
--- a/tensorflow_datasets/core/file_format_adapter.py
+++ b/tensorflow_datasets/core/file_format_adapter.py
@@ -386,6 +386,17 @@ def _item_to_tf_feature(item, key_name):
         "FeatureConnector should return a numpy array with the correct dtype "
         "instead of a Python list.".format(key_name)
     )
+  if isinstance(v, (six.binary_type, six.string_types)):
+    v = [tf.compat.as_bytes(v)]
+    return tf.train.Feature(bytes_list=tf.train.BytesList(value=v))
+  elif isinstance(v, (tuple, list)) and \
+          all(isinstance(x, (six.binary_type, six.string_types)) for x in v):
+    v = [tf.compat.as_bytes(x) for x in v]
+    return tf.train.Feature(bytes_list=tf.train.BytesList(value=v))
+  elif isinstance(v, np.ndarray) and \
+          (v.dtype.kind in ("U", "S") or v.dtype == object):  # binary or unicode
+    v = [tf.compat.as_bytes(x) for x in v.flatten()]
+    return tf.train.Feature(bytes_list=tf.train.BytesList(value=v))
 
   v = np.array(v).flatten()  # Convert v into a 1-d array
 
@@ -393,9 +404,6 @@ def _item_to_tf_feature(item, key_name):
     return tf.train.Feature(int64_list=tf.train.Int64List(value=v))
   elif np.issubdtype(v.dtype, np.floating):
     return tf.train.Feature(float_list=tf.train.FloatList(value=v))
-  elif v.dtype.kind in ("U", "S") or v.dtype == object:  # binary or unicode
-    v = [tf.compat.as_bytes(x) for x in v]
-    return tf.train.Feature(bytes_list=tf.train.BytesList(value=v))
   else:
     raise ValueError(
         "Value received: {}.\n"


Test patch:
diff --git a/tensorflow_datasets/core/file_format_adapter_test.py b/tensorflow_datasets/core/file_format_adapter_test.py
index f5e271d93..d333101eb 100644
--- a/tensorflow_datasets/core/file_format_adapter_test.py
+++ b/tensorflow_datasets/core/file_format_adapter_test.py
@@ -153,6 +153,9 @@ class TFRecordUtilsTest(testing.TestCase):
         "c2": np.array([2.0]),
         # Empty values supported when type is defined
         "d": np.array([], dtype=np.int32),
+        # Support for byte strings
+        "e": np.zeros(2, dtype=np.uint8).tobytes(),
+        "e2": [np.zeros(2, dtype=np.uint8).tobytes()] * 2,
     })
     feature = example.features.feature
     self.assertEqual([1], list(feature["a"].int64_list.value))
@@ -162,6 +165,8 @@ class TFRecordUtilsTest(testing.TestCase):
     self.assertEqual([2.0], list(feature["c"].float_list.value))
     self.assertEqual([2.0], list(feature["c2"].float_list.value))
     self.assertEqual([], list(feature["d"].int64_list.value))
+    self.assertEqual([b"\x00\x00"], list(feature["e"].bytes_list.value))
+    self.assertEqual([b"\x00\x00", b"\x00\x00"], list(feature["e2"].bytes_list.value))
 
     with self.assertRaisesWithPredicateMatch(ValueError, "received an empty"):
       # Raise error if an undefined empty value is given