Repository: pydata/xarray

Issue:
Unicode strings unexpectedly transformed to byte strings upon `open_dataset`
When I first create the dataset, all the metadata is stored as unicode strings (yay!):

```
<xarray.Dataset>
Dimensions:                       (cell: 53760, gene: 23438)
Coordinates:
  * gene                          (gene) object '0610005C13Rik' ...
    Uniquely mapped reads number  (cell) int64 1017682 634557 941828 1392029 ...
    Number of input reads         (cell) int64 1229254 730274 1075370 ...
    EXP_ID                        (cell) <U29 '170925_A00111_0066_AH3TKNDMXX' ...
    TAXON                         (cell) <U3 'mus' 'mus' 'mus' 'mus' 'mus' ...
    WELL_MAPPING                  (cell) <U9 'B000126' 'B000126' 'B000126' ...
    Lysis Plate Batch             (cell) <U32 '20' '20' '20' '20' '20' '20' ...
    dNTP.batch                    (cell) <U38 '457912' '457912' '457912' ...
    oligodT.order.no              (cell) <U32 '6/23/17 12757296' ...
    plate.type                    (cell) <U32 'Biorad HSP3901' ...
    preparation.site              (cell) <U32 'Biohub' 'Biohub' 'Biohub' ...
    date.prepared                 (cell) <U32 '07-06-17' '07-06-17' ...
    date.sorted                   (cell) <U6 '170707' '170707' '170707' ...
    tissue                        (cell) <U13 'Skin' 'Skin' 'Skin' 'Skin' ...
    subtissue                     (cell) <U32 'nan' 'nan' 'nan' 'nan' 'nan' ...
    mouse.id                      (cell) <U13 '3_39_F' '3_39_F' '3_39_F' ...
    FACS.selection                (cell) <U52 'Multiple' 'Multiple' ...
    nozzle.size                   (cell) <U32 '100' '100' '100' '100' '100' ...
    FACS.instument                (cell) <U32 'Sony SIM1' 'Sony SIM1' ...
    Experiment ID                 (cell) <U32 'exp22' 'exp22' 'exp22' ...
    Columns sorted                (cell) float64 nan nan nan nan nan nan nan ...
    Double check                  (cell) float64 nan nan nan nan nan nan nan ...
    Plate                         (cell) <U32 '1' '1' '1' '1' '1' '1' '1' ...
    Location                      (cell) <U32 'MACA20_3' 'MACA20_3' ...
    Comments                      (cell) <U32 'nan' 'nan' 'nan' 'nan' 'nan' ...
    mouse.age                     (cell) <U1 '3' '3' '3' '3' '3' '3' '3' '3' ...
    mouse.number                  (cell) <U32 '39' '39' '39' '39' '39' '39' ...
    mouse.sex                     (cell) <U1 'F' 'F' 'F' 'F' 'F' 'F' 'F' 'F' ...
  * cell                          (cell) object 'A17-B000126-3_39_F-1-1' ...
Data variables:
    counts                        (cell, gene) int64 0 0 0 0 442 0 0 0 0 0 0 ...
    log2                          (cell, gene) float64 0.0 0.0 0.0 0.0 8.791 ...
    log10                         (cell, gene) float64 0.0 0.0 0.0 0.0 2.646 ...
```

but then when I save using `to_netcdf` using the default arguments, then `xr.open_dataset` on the same dataset using default arguments, all of them get converted to byte strings:

```
<xarray.Dataset>
Dimensions:                       (cell: 53760, gene: 23438)
Coordinates:
  * cell                          (cell) |S24 b'A17-B000126-3_39_F-1-1' ...
  * gene                          (gene) |S22 b'0610005C13Rik' ...
Data variables:
    counts                        (cell, gene) int32 0 0 0 0 442 0 0 0 0 0 0 ...
    log2                          (cell, gene) float64 0.0 0.0 0.0 0.0 8.791 ...
    log10                         (cell, gene) float64 0.0 0.0 0.0 0.0 2.646 ...
    FACS.selection                (cell) |S52 b'Multiple' b'Multiple' ...
    dNTP.batch                    (cell) |S38 b'457912' b'457912' b'457912' ...
    EXP_ID                        (cell) |S29 b'170925_A00111_0066_AH3TKNDMXX' ...
    subtissue                     (cell) |S19 b'nan' b'nan' b'nan' b'nan' ...
    oligodT.order.no              (cell) |S17 b'6/23/17 12757296' ...
    plate.type                    (cell) |S14 b'Biorad HSP3901' ...
    tissue                        (cell) |S13 b'Skin' b'Skin' b'Skin' ...
    mouse.id                      (cell) |S13 b'3_39_F' b'3_39_F' b'3_39_F' ...
    FACS.instument                (cell) |S13 b'Sony SIM1' b'Sony SIM1' ...
    Comments                      (cell) |S11 b'nan' b'nan' b'nan' b'nan' ...
    WELL_MAPPING                  (cell) |S9 b'B000126' b'B000126' ...
    date.prepared                 (cell) |S9 b'07-06-17' b'07-06-17' ...
    Location                      (cell) |S9 b'MACA20_3' b'MACA20_3' ...
    preparation.site              (cell) |S8 b'Biohub' b'Biohub' b'Biohub' ...
    date.sorted                   (cell) |S6 b'170707' b'170707' b'170707' ...
    Experiment ID                 (cell) |S6 b'exp22' b'exp22' b'exp22' ...
    TAXON                         (cell) |S3 b'mus' b'mus' b'mus' b'mus' ...
    Lysis Plate Batch             (cell) |S3 b'20' b'20' b'20' b'20' b'20' ...
    nozzle.size                   (cell) |S3 b'100' b'100' b'100' b'100' ...
    Plate                         (cell) |S3 b'1' b'1' b'1' b'1' b'1' b'1' ...
    mouse.number                  (cell) |S3 b'39' b'39' b'39' b'39' b'39' ...
    Uniquely mapped reads number  (cell) int32 1017682 634557 941828 1392029 ...
    Number of input reads         (cell) int32 1229254 730274 1075370 ...
    Columns sorted                (cell) float64 nan nan nan nan nan nan nan ...
    Double check                  (cell) float64 nan nan nan nan nan nan nan ...
    mouse.age                     (cell) |S1 b'3' b'3' b'3' b'3' b'3' b'3' ...
    mouse.sex                     (cell) |S1 b'F' b'F' b'F' b'F' b'F' b'F' ...
```

So then things I expect like selecting on gene, e.g. `ds.sel(gene="Ins1")` don't work unless they're byte strings, i.e. `ds.sel(gene=b"Ins1")` works just fine. 

Do you know why this may be happening?

Discussion:

Patch:
diff --git a/doc/io.rst b/doc/io.rst
index a2be1ad3..9e17ffc0 100644
--- a/doc/io.rst
+++ b/doc/io.rst
@@ -270,6 +270,42 @@ converting ``NaN`` to ``-9999``, we would use
 ``encoding={'foo': {'dtype': 'int16', 'scale_factor': 0.1, '_FillValue': -9999}}``.
 Compression and decompression with such discretization is extremely fast.
 
+.. _io.string-encoding:
+
+String encoding
+...............
+
+xarray can write unicode strings to netCDF files in two ways:
+
+- As variable length strings. This is only supported on netCDF4 (HDF5) files.
+- By encoding strings into bytes, and writing encoded bytes as a character
+  array. The default encoding is UTF-8.
+
+By default, we use variable length strings for compatible files and fall-back
+to using encoded character arrays. Character arrays can be selected even for
+netCDF4 files by setting the ``dtype`` field in ``encoding`` to ``S1``
+(corresponding to NumPy's single-character bytes dtype).
+
+If character arrays are used, the string encoding that was used is stored on
+disk in the ``_Encoding`` attribute, which matches an ad-hoc convention
+`adopted by the netCDF4-Python library <https://github.com/Unidata/netcdf4-python/pull/665>`_.
+At the time of this writing (October 2017), a standard convention for indicating
+string encoding for character arrays in netCDF files was
+`still under discussion <https://github.com/Unidata/netcdf-c/issues/402>`_.
+Technically, you can use
+`any string encoding recognized by Python <https://docs.python.org/3/library/codecs.html#standard-encodings>`_ if you feel the need to deviate from UTF-8,
+by setting the ``_Encoding`` field in ``encoding``. But
+`we don't recommend it<http://utf8everywhere.org/>`_.
+
+.. warning::
+
+  Missing values in bytes or unicode string arrays (represented by ``NaN`` in
+  xarray) are currently written to disk as empty strings ``''``. This means
+  missing values will not be restored when data is loaded from disk.
+  This behavior is likely to change in the future (:issue:`1647`).
+  Unfortunately, explicitly setting a ``_FillValue`` for string arrays to handle
+  missing values doesn't work yet either, though we also hope to fix this in the
+  future.
 
 Chunk based compression
 .......................
diff --git a/doc/whats-new.rst b/doc/whats-new.rst
index d423d8d2..a27bd1a2 100644
--- a/doc/whats-new.rst
+++ b/doc/whats-new.rst
@@ -73,6 +73,13 @@ Breaking changes
   produce a warning encouraging users to adopt the new syntax.
   By `Daniel Rothenberg <https://github.com/darothen>`_.
 
+- Unicode strings (``str`` on Python 3) are now round-tripped successfully even
+  when written as character arrays (e.g., as netCDF3 files or when using
+  ``engine='scipy'``) (:issue:`1638`). This is controlled by the ``_Encoding``
+  attribute convention, which is also understood directly by the netCDF4-Python
+  interface. See :ref:`io.string-encoding` for full details.
+  By `Stephan Hoyer <https://github.com/shoyer>`_.
+
 - ``repr`` and the Jupyter Notebook won't automatically compute dask variables.
   Datasets loaded with ``open_dataset`` won't automatically read coords from
   disk when calling ``repr`` (:issue:`1522`).
diff --git a/xarray/backends/common.py b/xarray/backends/common.py
index 6a7c5e4b..1c63c368 100644
--- a/xarray/backends/common.py
+++ b/xarray/backends/common.py
@@ -241,6 +241,7 @@ class AbstractWritableDataStore(AbstractDataStore):
 
 
 class WritableCFDataStore(AbstractWritableDataStore):
+
     def store(self, variables, attributes, *args, **kwargs):
         # All NetCDF files get CF encoded by default, without this attempting
         # to write times, for example, would fail.
diff --git a/xarray/backends/h5netcdf_.py b/xarray/backends/h5netcdf_.py
index 8f790be6..5bd4fa6c 100644
--- a/xarray/backends/h5netcdf_.py
+++ b/xarray/backends/h5netcdf_.py
@@ -131,14 +131,21 @@ class H5NetCDFStore(WritableCFDataStore, DataStorePickleMixin):
 
         attrs = variable.attrs.copy()
         variable, dtype = _nc4_values_and_dtype(variable)
-        if dtype is str:
-            dtype = h5py.special_dtype(vlen=unicode_type)
 
         self.set_necessary_dimensions(variable, unlimited_dims=unlimited_dims)
 
         fill_value = attrs.pop('_FillValue', None)
-        if fill_value in ['\x00']:
-            fill_value = None
+        if dtype is str and fill_value is not None:
+            raise NotImplementedError(
+                'h5netcdf does not yet support setting a fill value for '
+                'variable-length strings '
+                '(https://github.com/shoyer/h5netcdf/issues/37). '
+                "Either remove '_FillValue' from encoding on variable %r "
+                "or set {'dtype': 'S1'} in encoding to use the fixed width "
+                'NC_CHAR type.' % name)
+
+        if dtype is str:
+            dtype = h5py.special_dtype(vlen=unicode_type)
 
         encoding = _extract_h5nc_encoding(variable,
                                           raise_on_invalid=check_encoding)
diff --git a/xarray/backends/netCDF4_.py b/xarray/backends/netCDF4_.py
index f2c3a66c..52785f30 100644
--- a/xarray/backends/netCDF4_.py
+++ b/xarray/backends/netCDF4_.py
@@ -6,18 +6,18 @@ import operator
 
 import numpy as np
 
+from .. import conventions
 from .. import Variable
 from ..conventions import pop_to
 from ..core import indexing
 from ..core.utils import (FrozenOrderedDict, NdimSizeLenMixin,
                           DunderArrayMixin, close_on_error,
                           is_remote_uri)
-from ..core.pycompat import iteritems, basestring, OrderedDict, PY3
+from ..core.pycompat import iteritems, basestring, OrderedDict, PY3, suppress
 
 from .common import (WritableCFDataStore, robust_getitem,
                      DataStorePickleMixin, find_root)
-from .netcdf3 import (encode_nc3_attr_value, encode_nc3_variable,
-                      maybe_convert_to_char_array)
+from .netcdf3 import (encode_nc3_attr_value, encode_nc3_variable)
 
 # This lookup table maps from dtype.byteorder to a readable endian
 # string used by netCDF4.
@@ -72,25 +72,16 @@ class NetCDF4ArrayWrapper(BaseNetCDF4Array):
                     msg += '\n\nOriginal traceback:\n' + traceback.format_exc()
                 raise IndexError(msg)
 
-        if self.ndim == 0:
-            # work around for netCDF4-python's broken handling of 0-d
-            # arrays (slicing them always returns a 1-dimensional array):
-            # https://github.com/Unidata/netcdf4-python/pull/220
-            data = np.asscalar(data)
         return data
 
 
 def _nc4_values_and_dtype(var):
     if var.dtype.kind == 'U':
-        # this entire clause should not be necessary with netCDF4>=1.0.9
-        if len(var) > 0:
-            var = var.astype('O')
         dtype = str
     elif var.dtype.kind == 'S':
         # use character arrays instead of unicode, because unicode support in
         # netCDF4 is still rather buggy
-        data, dims = maybe_convert_to_char_array(var.data, var.dims)
-        var = Variable(dims, data, var.attrs, var.encoding)
+        var = conventions.maybe_encode_as_char_array(var)
         dtype = var.dtype
     elif var.dtype.kind in ['i', 'u', 'f', 'c']:
         dtype = var.dtype
@@ -189,15 +180,27 @@ def _open_netcdf4_group(filename, mode, group=None, **kwargs):
     with close_on_error(ds):
         ds = _nc4_group(ds, group, mode)
 
-    _disable_mask_and_scale(ds)
+    _disable_auto_decode_group(ds)
 
     return ds
 
 
-def _disable_mask_and_scale(ds):
+def _disable_auto_decode_variable(var):
+    """Disable automatic decoding on a netCDF4.Variable.
+
+    We handle these types of decoding ourselves.
+    """
+    var.set_auto_maskandscale(False)
+
+    # only added in netCDF4-python v1.2.8
+    with suppress(AttributeError):
+        var.set_auto_chartostring(False)
+
+
+def _disable_auto_decode_group(ds):
+    """Disable automatic decoding on all variables in a netCDF4.Group."""
     for var in ds.variables.values():
-        # we handle masking and scaling ourselves
-        var.set_auto_maskandscale(False)
+        _disable_auto_decode_variable(var)
 
 
 class NetCDF4DataStore(WritableCFDataStore, DataStorePickleMixin):
@@ -211,7 +214,7 @@ class NetCDF4DataStore(WritableCFDataStore, DataStorePickleMixin):
         if autoclose and opener is None:
             raise ValueError('autoclose requires an opener')
 
-        _disable_mask_and_scale(netcdf4_dataset)
+        _disable_auto_decode_group(netcdf4_dataset)
 
         self.ds = netcdf4_dataset
         self._autoclose = autoclose
@@ -313,8 +316,6 @@ class NetCDF4DataStore(WritableCFDataStore, DataStorePickleMixin):
 
     def prepare_variable(self, name, variable, check_encoding=False,
                          unlimited_dims=None):
-        attrs = variable.attrs.copy()
-
         variable = _force_native_endianness(variable)
 
         if self.format == 'NETCDF4':
@@ -325,11 +326,18 @@ class NetCDF4DataStore(WritableCFDataStore, DataStorePickleMixin):
 
         self.set_necessary_dimensions(variable, unlimited_dims=unlimited_dims)
 
+        attrs = variable.attrs.copy()
+
         fill_value = attrs.pop('_FillValue', None)
-        if fill_value in ['', '\x00']:
-            # these are equivalent to the default FillValue, but netCDF4
-            # doesn't like setting fill_value to an empty string
-            fill_value = None
+
+        if datatype is str and fill_value is not None:
+            raise NotImplementedError(
+                'netCDF4 does not yet support setting a fill value for '
+                'variable-length strings '
+                '(https://github.com/Unidata/netcdf4-python/issues/730). '
+                "Either remove '_FillValue' from encoding on variable %r "
+                "or set {'dtype': 'S1'} in encoding to use the fixed width "
+                'NC_CHAR type.' % name)
 
         encoding = _extract_nc4_variable_encoding(
             variable, raise_on_invalid=check_encoding)
@@ -346,7 +354,7 @@ class NetCDF4DataStore(WritableCFDataStore, DataStorePickleMixin):
             endian='native',
             least_significant_digit=encoding.get('least_significant_digit'),
             fill_value=fill_value)
-        nc4_var.set_auto_maskandscale(False)
+        _disable_auto_decode_variable(nc4_var)
 
         for k, v in iteritems(attrs):
             # set attributes one-by-one since netCDF4<1.0.10 can't handle
diff --git a/xarray/backends/netcdf3.py b/xarray/backends/netcdf3.py
index 24d76a80..7194e061 100644
--- a/xarray/backends/netcdf3.py
+++ b/xarray/backends/netcdf3.py
@@ -24,15 +24,16 @@ _reserved_names = set(['byte', 'char', 'short', 'ushort', 'int', 'uint',
 # coerced instead as indicated by the "coerce_nc3_dtype" function
 _nc3_dtype_coercions = {'int64': 'int32', 'bool': 'int8'}
 
+# encode all strings as UTF-8
+STRING_ENCODING = 'utf-8'
+
 
 def coerce_nc3_dtype(arr):
     """Coerce an array to a data type that can be stored in a netCDF-3 file
 
     This function performs the following dtype conversions:
         int64 -> int32
-        float64 -> float32
         bool -> int8
-        unicode -> string
 
     Data is checked for equality, or equivalence (non-NaN values) with
     `np.allclose` with the default keyword arguments.
@@ -42,29 +43,18 @@ def coerce_nc3_dtype(arr):
         new_dtype = _nc3_dtype_coercions[dtype]
         # TODO: raise a warning whenever casting the data-type instead?
         cast_arr = arr.astype(new_dtype)
-        if ((('int' in dtype or 'U' in dtype) and
-                not (cast_arr == arr).all()) or
-                ('float' in dtype and
-                    not duck_array_ops.allclose_or_equiv(cast_arr, arr))):
+        if not (cast_arr == arr).all():
             raise ValueError('could not safely cast array from dtype %s to %s'
                              % (dtype, new_dtype))
         arr = cast_arr
-    elif arr.dtype.kind == 'U':
-        arr = np.core.defchararray.encode(arr, 'utf-8')
     return arr
 
 
-def maybe_convert_to_char_array(data, dims):
-    if data.dtype.kind == 'S' and data.dtype.itemsize > 1:
-        data = conventions.string_to_char(data)
-        dims = dims + ('string%s' % data.shape[-1],)
-    return data, dims
-
-
 def encode_nc3_attr_value(value):
-    if isinstance(value, basestring):
-        if not isinstance(value, unicode_type):
-            value = value.decode('utf-8')
+    if isinstance(value, bytes):
+        pass
+    elif isinstance(value, unicode_type):
+        value = value.encode(STRING_ENCODING)
     else:
         value = coerce_nc3_dtype(np.atleast_1d(value))
         if value.ndim > 1:
@@ -78,10 +68,10 @@ def encode_nc3_attrs(attrs):
 
 
 def encode_nc3_variable(var):
+    var = conventions.maybe_encode_as_char_array(var)
     data = coerce_nc3_dtype(var.data)
-    data, dims = maybe_convert_to_char_array(data, var.dims)
     attrs = encode_nc3_attrs(var.attrs)
-    return Variable(dims, data, attrs, var.encoding)
+    return Variable(var.dims, data, attrs, var.encoding)
 
 
 def _isalnumMUTF8(c):
diff --git a/xarray/conventions.py b/xarray/conventions.py
index 6c4d63bc..e78374b8 100644
--- a/xarray/conventions.py
+++ b/xarray/conventions.py
@@ -447,12 +447,11 @@ class DecodedCFTimedeltaArray(utils.NDArrayMixin):
         return decode_cf_timedelta(self.array[key], units=self.units)
 
 
-class CharToStringArray(utils.NDArrayMixin):
+class StackedBytesArray(utils.NDArrayMixin):
     """Wrapper around array-like objects to create a new indexable object where
-    values, when accessed, are automatically concatenated along the last
-    dimension.
+    values, when accessed, are automatically stacked along the last dimension.
 
-    >>> CharToStringArray(np.array(['a', 'b', 'c']))[:]
+    >>> StackedBytesArray(np.array(['a', 'b', 'c']))[:]
     array('abc',
           dtype='|S3')
     """
@@ -463,6 +462,9 @@ class CharToStringArray(utils.NDArrayMixin):
         array : array-like
             Original array of values to wrap.
         """
+        if array.dtype != 'S1':
+            raise ValueError(
+                "can only use StackedBytesArray if argument has dtype='S1'")
         self.array = array
 
     @property
@@ -474,25 +476,60 @@ class CharToStringArray(utils.NDArrayMixin):
         return self.array.shape[:-1]
 
     def __str__(self):
+        # TODO(shoyer): figure out why we need this special case?
         if self.ndim == 0:
-            # always return a unicode str if it's a single item for py3 compat
-            return self[...].item().decode('utf-8')
+            return str(self[...].item())
         else:
             return repr(self)
 
     def __repr__(self):
-        return '%s(%r)' % (type(self).__name__, self.array)
+        return ('%s(%r)' % (type(self).__name__, self.array))
 
     def __getitem__(self, key):
-        if self.array.ndim == 0:
-            values = self.array[key]
+        # require slicing the last dimension completely
+        key = indexing.expanded_indexer(key, self.array.ndim)
+        if key[-1] != slice(None):
+            raise IndexError('too many indices')
+        return char_to_bytes(self.array[key])
+
+
+class BytesToStringArray(utils.NDArrayMixin):
+    """Wrapper that decodes bytes to unicode when values are read.
+
+    >>> BytesToStringArray(np.array([b'abc']))[:]
+    array(['abc'],
+          dtype=object)
+    """
+    def __init__(self, array, encoding='utf-8'):
+        """
+        Parameters
+        ----------
+        array : array-like
+            Original array of values to wrap.
+        encoding : str
+            String encoding to use.
+        """
+        self.array = array
+        self.encoding = encoding
+
+    @property
+    def dtype(self):
+        # variable length string
+        return np.dtype(object)
+
+    def __str__(self):
+        # TODO(shoyer): figure out why we need this special case?
+        if self.ndim == 0:
+            return str(self[...].item())
         else:
-            # require slicing the last dimension completely
-            key = indexing.expanded_indexer(key, self.array.ndim)
-            if key[-1] != slice(None):
-                raise IndexError('too many indices')
-            values = char_to_string(self.array[key])
-        return values
+            return repr(self)
+
+    def __repr__(self):
+        return ('%s(%r, encoding=%r)'
+                % (type(self).__name__, self.array, self.encoding))
+
+    def __getitem__(self, key):
+        return decode_bytes_array(self.array[key], self.encoding)
 
 
 class NativeEndiannessArray(utils.NDArrayMixin):
@@ -580,35 +617,65 @@ class UnsignedIntTypeArray(utils.NDArrayMixin):
         return np.asarray(self.array[key], dtype=self.dtype)
 
 
-def string_to_char(arr):
+def bytes_to_char(arr):
     """Like netCDF4.stringtochar, but faster and more flexible.
     """
     # ensure the array is contiguous
     arr = np.array(arr, copy=False, order='C')
     kind = arr.dtype.kind
     if kind not in ['U', 'S']:
-        raise ValueError('argument must be a string')
+        raise ValueError('argument must be a string array')
     return arr.reshape(arr.shape + (1,)).view(kind + '1')
 
 
-def char_to_string(arr):
+def char_to_bytes(arr):
     """Like netCDF4.chartostring, but faster and more flexible.
     """
     # based on: http://stackoverflow.com/a/10984878/809705
     arr = np.array(arr, copy=False, order='C')
+
     kind = arr.dtype.kind
     if kind not in ['U', 'S']:
-        raise ValueError('argument must be a string')
-    return arr.view(kind + str(arr.shape[-1]))[..., 0]
+        raise ValueError('argument must be a string array')
+
+    if not arr.ndim:
+        # no dimension to concatenate along
+        return arr
+
+    size = arr.shape[-1]
+    if not size:
+        # can't make an S0 dtype
+        return np.zeros(arr.shape[:-1], dtype=kind)
+
+    dtype = kind + str(size)
+    return arr.view(dtype).reshape(arr.shape[:-1])
+
+
+def decode_bytes_array(bytes_array, encoding='utf-8'):
+    # This is faster than using np.char.decode() or np.vectorize()
+    bytes_array = np.asarray(bytes_array)
+    decoded = [x.decode(encoding) for x in bytes_array.ravel()]
+    return np.array(decoded, dtype=object).reshape(bytes_array.shape)
 
 
-def safe_setitem(dest, key, value):
+def encode_string_array(string_array, encoding='utf-8'):
+    string_array = np.asarray(string_array)
+    encoded = [x.encode(encoding) for x in string_array.ravel()]
+    return np.array(encoded, dtype=bytes).reshape(string_array.shape)
+
+
+def safe_setitem(dest, key, value, name=None):
     if key in dest:
-        raise ValueError('Failed hard to prevent overwriting key %r' % key)
+        var_str = ' on variable {!r}'.format(name) if name else ''
+        raise ValueError(
+            'failed to prevent overwriting existing key {} in attrs{}. '
+            'This is probably an encoding field used by xarray to describe '
+            'how a variable is serialized. To proceed, remove this key from '
+            "the variable's attributes manually.".format(key, var_str))
     dest[key] = value
 
 
-def pop_to(source, dest, key, default=None):
+def pop_to(source, dest, key, name=None):
     """
     A convenience function which pops a key k from source to dest.
     None values are not passed on.  If k already exists in dest an
@@ -616,7 +683,7 @@ def pop_to(source, dest, key, default=None):
     """
     value = source.pop(key, None)
     if value is not None:
-        safe_setitem(dest, key, value)
+        safe_setitem(dest, key, value, name=name)
     return value
 
 
@@ -624,45 +691,45 @@ def _var_as_tuple(var):
     return var.dims, var.data, var.attrs.copy(), var.encoding.copy()
 
 
-def maybe_encode_datetime(var):
+def maybe_encode_datetime(var, name=None):
     if np.issubdtype(var.dtype, np.datetime64):
         dims, data, attrs, encoding = _var_as_tuple(var)
         (data, units, calendar) = encode_cf_datetime(
             data, encoding.pop('units', None), encoding.pop('calendar', None))
-        safe_setitem(attrs, 'units', units)
-        safe_setitem(attrs, 'calendar', calendar)
+        safe_setitem(attrs, 'units', units, name=name)
+        safe_setitem(attrs, 'calendar', calendar, name=name)
         var = Variable(dims, data, attrs, encoding)
     return var
 
 
-def maybe_encode_timedelta(var):
+def maybe_encode_timedelta(var, name=None):
     if np.issubdtype(var.dtype, np.timedelta64):
         dims, data, attrs, encoding = _var_as_tuple(var)
         data, units = encode_cf_timedelta(
             data, encoding.pop('units', None))
-        safe_setitem(attrs, 'units', units)
+        safe_setitem(attrs, 'units', units, name=name)
         var = Variable(dims, data, attrs, encoding)
     return var
 
 
-def maybe_encode_offset_and_scale(var, needs_copy=True):
+def maybe_encode_offset_and_scale(var, needs_copy=True, name=None):
     if any(k in var.encoding for k in ['add_offset', 'scale_factor']):
         dims, data, attrs, encoding = _var_as_tuple(var)
         data = data.astype(dtype=float, copy=needs_copy)
         needs_copy = False
         if 'add_offset' in encoding:
-            data -= pop_to(encoding, attrs, 'add_offset')
+            data -= pop_to(encoding, attrs, 'add_offset', name=name)
         if 'scale_factor' in encoding:
-            data /= pop_to(encoding, attrs, 'scale_factor')
+            data /= pop_to(encoding, attrs, 'scale_factor', name=name)
         var = Variable(dims, data, attrs, encoding)
     return var, needs_copy
 
 
-def maybe_encode_fill_value(var, needs_copy=True):
+def maybe_encode_fill_value(var, needs_copy=True, name=None):
     # replace NaN with the fill value
     if var.encoding.get('_FillValue') is not None:
         dims, data, attrs, encoding = _var_as_tuple(var)
-        fill_value = pop_to(encoding, attrs, '_FillValue')
+        fill_value = pop_to(encoding, attrs, '_FillValue', name=name)
         if not pd.isnull(fill_value):
             data = ops.fillna(data, fill_value)
             needs_copy = False
@@ -670,8 +737,33 @@ def maybe_encode_fill_value(var, needs_copy=True):
     return var, needs_copy
 
 
-def maybe_encode_dtype(var, name=None):
-    if 'dtype' in var.encoding:
+def maybe_encode_as_char_array(var, name=None):
+    if var.dtype.kind in {'S', 'U'}:
+        dims, data, attrs, encoding = _var_as_tuple(var)
+        if data.dtype.kind == 'U':
+            string_encoding = encoding.pop('_Encoding', 'utf-8')
+            safe_setitem(attrs, '_Encoding', string_encoding, name=name)
+            data = encode_string_array(data, string_encoding)
+
+        if data.dtype.itemsize > 1:
+            data = bytes_to_char(data)
+            dims = dims + ('string%s' % data.shape[-1],)
+
+        var = Variable(dims, data, attrs, encoding)
+    return var
+
+
+def maybe_encode_string_dtype(var, name=None):
+    # need to apply after ensure_dtype_not_object()
+    if 'dtype' in var.encoding and var.encoding['dtype'] == 'S1':
+        assert var.dtype.kind in {'S', 'U'}
+        var = maybe_encode_as_char_array(var, name=name)
+        del var.encoding['dtype']
+    return var
+
+
+def maybe_encode_nonstring_dtype(var, name=None):
+    if 'dtype' in var.encoding and var.encoding['dtype'] != 'S1':
         dims, data, attrs, encoding = _var_as_tuple(var)
         dtype = np.dtype(encoding.pop('dtype'))
         if dtype != var.dtype:
@@ -684,16 +776,12 @@ def maybe_encode_dtype(var, name=None):
                                   RuntimeWarning, stacklevel=3)
                 data = duck_array_ops.around(data)[...]
                 if encoding.get('_Unsigned', False):
-                    signed_dtype = 'i%s' % dtype.itemsize
+                    signed_dtype = np.dtype('i%s' % dtype.itemsize)
                     if '_FillValue' in var.attrs:
-                        old_fill = np.asarray(attrs['_FillValue'])
-                        new_fill = old_fill.astype(signed_dtype)
+                        new_fill = signed_dtype.type(attrs['_FillValue'])
                         attrs['_FillValue'] = new_fill
                     data = data.astype(signed_dtype)
                     pop_to(encoding, attrs, '_Unsigned')
-            if dtype == 'S1' and data.dtype != 'S1':
-                data = string_to_char(np.asarray(data, 'S'))
-                dims = dims + ('string%s' % data.shape[-1],)
             data = data.astype(dtype=dtype)
         var = Variable(dims, data, attrs, encoding)
     return var
@@ -704,7 +792,7 @@ def maybe_default_fill_value(var):
     if ('_FillValue' not in var.attrs and
             '_FillValue' not in var.encoding and
             np.issubdtype(var.dtype, np.floating)):
-        var.attrs['_FillValue'] = np.nan
+        var.attrs['_FillValue'] = var.dtype.type(np.nan)
     return var
 
 
@@ -718,7 +806,7 @@ def maybe_encode_bools(var):
     return var
 
 
-def _infer_dtype(array):
+def _infer_dtype(array, name=None):
     """Given an object array with no missing values, infer its dtype from its
     first element
     """
@@ -731,12 +819,13 @@ def _infer_dtype(array):
             # the length of their first element
             dtype = np.dtype(dtype.kind)
         elif dtype.kind == 'O':
-            raise ValueError('unable to infer dtype; xarray cannot '
-                             'serialize arbitrary Python objects')
+            raise ValueError('unable to infer dtype on variable {!r}; xarray '
+                             'cannot serialize arbitrary Python objects'
+                             .format(name))
     return dtype
 
 
-def ensure_dtype_not_object(var):
+def ensure_dtype_not_object(var, name=None):
     # TODO: move this from conventions to backends? (it's not CF related)
     if var.dtype.kind == 'O':
         dims, data, attrs, encoding = _var_as_tuple(var)
@@ -744,24 +833,25 @@ def ensure_dtype_not_object(var):
         if missing.any():
             # nb. this will fail for dask.array data
             non_missing_values = data[~missing]
-            inferred_dtype = _infer_dtype(non_missing_values)
-
-            if inferred_dtype.kind in ['S', 'U']:
-                # There is no safe bit-pattern for NA in typical binary string
-                # formats, we so can't set a fill_value. Unfortunately, this
-                # means we won't be able to restore string arrays with missing
-                # values.
-                fill_value = ''
+            inferred_dtype = _infer_dtype(non_missing_values, name)
+
+            # There is no safe bit-pattern for NA in typical binary string
+            # formats, we so can't set a fill_value. Unfortunately, this means
+            # we can't distinguish between missing values and empty strings.
+            if inferred_dtype.kind == 'S':
+                fill_value = b''
+            elif inferred_dtype.kind == 'U':
+                fill_value = u''
             else:
                 # insist on using float for numeric values
                 if not np.issubdtype(inferred_dtype, float):
                     inferred_dtype = np.dtype(float)
-                fill_value = np.nan
+                fill_value = inferred_dtype.type(np.nan)
 
             data = np.array(data, dtype=inferred_dtype, copy=True)
             data[missing] = fill_value
         else:
-            data = data.astype(dtype=_infer_dtype(data))
+            data = data.astype(dtype=_infer_dtype(data, name))
         var = Variable(dims, data, attrs, encoding)
     return var
 
@@ -786,19 +876,21 @@ def encode_cf_variable(var, needs_copy=True, name=None):
     out : xarray.Variable
         A variable which has been encoded as described above.
     """
-    var = maybe_encode_datetime(var)
-    var = maybe_encode_timedelta(var)
-    var, needs_copy = maybe_encode_offset_and_scale(var, needs_copy)
-    var, needs_copy = maybe_encode_fill_value(var, needs_copy)
-    var = maybe_encode_dtype(var, name)
+    var = maybe_encode_datetime(var, name=name)
+    var = maybe_encode_timedelta(var, name=name)
+    var, needs_copy = maybe_encode_offset_and_scale(var, needs_copy, name=name)
+    var, needs_copy = maybe_encode_fill_value(var, needs_copy, name=name)
+    var = maybe_encode_nonstring_dtype(var, name=name)
     var = maybe_default_fill_value(var)
     var = maybe_encode_bools(var)
-    var = ensure_dtype_not_object(var)
+    var = ensure_dtype_not_object(var, name=name)
+    var = maybe_encode_string_dtype(var, name=name)
     return var
 
 
-def decode_cf_variable(var, concat_characters=True, mask_and_scale=True,
-                       decode_times=True, decode_endianness=True):
+def decode_cf_variable(name, var, concat_characters=True, mask_and_scale=True,
+                       decode_times=True, decode_endianness=True,
+                       stack_char_dim=True):
     """
     Decodes a variable which may hold CF encoded information.
 
@@ -809,6 +901,8 @@ def decode_cf_variable(var, concat_characters=True, mask_and_scale=True,
 
     Parameters
     ----------
+    name: str
+        Name of the variable. Used for better error messages.
     var : Variable
         A variable holding potentially CF encoded information.
     concat_characters : bool
@@ -822,11 +916,15 @@ def decode_cf_variable(var, concat_characters=True, mask_and_scale=True,
         Decode cf times ('hours since 2000-01-01') to np.datetime64.
     decode_endianness : bool
         Decode arrays from non-native to native endianness.
+    stack_char_dim : bool
+        Whether to stack characters into bytes along the last dimension of this
+        array. Passed as an argument because we need to look at the full
+        dataset to figure out if this is appropriate.
 
     Returns
     -------
     out : Variable
-        A variable holding the decoded equivalent of var
+        A variable holding the decoded equivalent of var.
     """
     # use _data instead of data so as not to trigger loading data
     var = as_variable(var)
@@ -837,19 +935,22 @@ def decode_cf_variable(var, concat_characters=True, mask_and_scale=True,
 
     original_dtype = data.dtype
 
-    if concat_characters:
-        if data.dtype.kind == 'S' and data.dtype.itemsize == 1:
+    if concat_characters and data.dtype.kind == 'S':
+        if stack_char_dim:
             dimensions = dimensions[:-1]
-            data = CharToStringArray(data)
+            data = StackedBytesArray(data)
+
+        string_encoding = pop_to(attributes, encoding, '_Encoding')
+        if string_encoding is not None:
+            data = BytesToStringArray(data, string_encoding)
 
-    pop_to(attributes, encoding, '_Unsigned')
-    is_unsigned = encoding.get('_Unsigned', False)
-    if is_unsigned and mask_and_scale:
+    unsigned = pop_to(attributes, encoding, '_Unsigned')
+    if unsigned and mask_and_scale:
         if data.dtype.kind == 'i':
             data = UnsignedIntTypeArray(data)
         else:
-            warnings.warn("variable has _Unsigned attribute but is not "
-                          "of integer type. Ignoring attribute.",
+            warnings.warn("variable %r has _Unsigned attribute but is not "
+                          "of integer type. Ignoring attribute." % name,
                           RuntimeWarning, stacklevel=3)
 
     if mask_and_scale:
@@ -860,34 +961,43 @@ def decode_cf_variable(var, concat_characters=True, mask_and_scale=True,
                 not utils.equivalent(attributes['_FillValue'],
                                      attributes['missing_value'])):
                 raise ValueError("Conflicting _FillValue and missing_value "
-                                 "attributes on a variable: {} vs. {}\n\n"
+                                 "attributes on a variable {!r}: {} vs. {}\n\n"
                                  "Consider opening the offending dataset "
                                  "using decode_cf=False, correcting the "
                                  "attributes and decoding explicitly using "
                                  "xarray.decode_cf()."
-                                 .format(attributes['_FillValue'],
+                                 .format(name, attributes['_FillValue'],
                                          attributes['missing_value']))
             attributes['_FillValue'] = attributes.pop('missing_value')
-        fill_value = np.array(pop_to(attributes, encoding, '_FillValue'))
-        if fill_value.size > 1:
-            warnings.warn("variable has multiple fill values {0}, decoding "
-                          "all values to NaN.".format(str(fill_value)),
+
+        fill_value = pop_to(attributes, encoding, '_FillValue')
+        if isinstance(fill_value, np.ndarray) and fill_value.size > 1:
+            warnings.warn("variable {!r} has multiple fill values {}, "
+                          "decoding all values to NaN."
+                          .format(name, fill_value),
                           RuntimeWarning, stacklevel=3)
+
         scale_factor = pop_to(attributes, encoding, 'scale_factor')
         add_offset = pop_to(attributes, encoding, 'add_offset')
         has_fill = (fill_value is not None and
                     not np.any(pd.isnull(fill_value)))
         if (has_fill or scale_factor is not None or add_offset is not None):
-            if fill_value.dtype.kind in ['U', 'S']:
+            if has_fill and np.array(fill_value).dtype.kind in ['U', 'S', 'O']:
+                if string_encoding is not None:
+                    raise NotImplementedError(
+                        'variable %r has a _FillValue specified, but '
+                        '_FillValue is yet supported on unicode strings: '
+                        'https://github.com/pydata/xarray/issues/1647')
                 dtype = object
             else:
+                # According to the CF spec, the fill value is of the same
+                # type as its variable, i.e. its storage format on disk.
+                # This handles the case where the fill_value also needs to be
+                # converted to its unsigned value.
+                if has_fill:
+                    fill_value = data.dtype.type(fill_value)
                 dtype = float
-            # According to the CF spec, the fill value is of the same
-            # type as its variable, i.e. its storage format on disk.
-            # This handles the case where the fill_value also needs to be
-            # converted to its unsigned value.
-            if has_fill:
-                fill_value = np.asarray(fill_value, dtype=data.dtype)
+
             data = MaskedAndScaledArray(data, fill_value, scale_factor,
                                         add_offset, dtype)
 
@@ -909,7 +1019,8 @@ def decode_cf_variable(var, concat_characters=True, mask_and_scale=True,
 
     if 'dtype' in encoding:
         if original_dtype != encoding['dtype']:
-            warnings.warn("CF decoding is overwriting dtype")
+            warnings.warn("CF decoding is overwriting dtype on variable {!r}"
+                          .format(name))
     else:
         encoding['dtype'] = original_dtype
 
@@ -955,11 +1066,12 @@ def decode_cf_variables(variables, attributes, concat_characters=True,
     for k, v in iteritems(variables):
         if k in drop_variables:
             continue
-        concat = (concat_characters and v.dtype.kind == 'S' and v.ndim > 0 and
-                  stackable(v.dims[-1]))
+        stack_char_dim = (concat_characters and v.dtype == 'S1' and
+                          v.ndim > 0 and stackable(v.dims[-1]))
         new_vars[k] = decode_cf_variable(
-            v, concat_characters=concat, mask_and_scale=mask_and_scale,
-            decode_times=decode_times)
+            k, v, concat_characters=concat_characters,
+            mask_and_scale=mask_and_scale, decode_times=decode_times,
+            stack_char_dim=stack_char_dim)
         if decode_coords:
             var_attrs = new_vars[k].attrs
             if 'coordinates' in var_attrs:


Test patch:
diff --git a/xarray/testing.py b/xarray/testing.py
index 23b06f98..69270848 100644
--- a/xarray/testing.py
+++ b/xarray/testing.py
@@ -19,7 +19,7 @@ def _data_allclose_or_equiv(arr1, arr2, rtol=1e-05, atol=1e-08,
     if any(arr.dtype.kind == 'S' for arr in [arr1, arr2]) and decode_bytes:
         arr1 = _decode_string_data(arr1)
         arr2 = _decode_string_data(arr2)
-    exact_dtypes = ['M', 'm', 'O', 'U']
+    exact_dtypes = ['M', 'm', 'O', 'S', 'U']
     if any(arr.dtype.kind in exact_dtypes for arr in [arr1, arr2]):
         return duck_array_ops.array_equiv(arr1, arr2)
     else:
@@ -49,7 +49,7 @@ def assert_equal(a, b):
     numpy.testing.assert_array_equal
     """
     import xarray as xr
-    ___tracebackhide__ = True  # noqa: F841
+    __tracebackhide__ = True  # noqa: F841
     assert type(a) == type(b)
     if isinstance(a, (xr.Variable, xr.DataArray, xr.Dataset)):
         assert a.equals(b), '{}\n{}'.format(a, b)
@@ -76,7 +76,7 @@ def assert_identical(a, b):
     assert_equal, assert_allclose, Dataset.equals, DataArray.equals
     """
     import xarray as xr
-    ___tracebackhide__ = True  # noqa: F841
+    __tracebackhide__ = True  # noqa: F841
     assert type(a) == type(b)
     if isinstance(a, xr.DataArray):
         assert a.name == b.name
@@ -114,7 +114,7 @@ def assert_allclose(a, b, rtol=1e-05, atol=1e-08, decode_bytes=True):
     assert_identical, assert_equal, numpy.testing.assert_allclose
     """
     import xarray as xr
-    ___tracebackhide__ = True  # noqa: F841
+    __tracebackhide__ = True  # noqa: F841
     assert type(a) == type(b)
     kwargs = dict(rtol=rtol, atol=atol, decode_bytes=decode_bytes)
     if isinstance(a, xr.Variable):
diff --git a/xarray/tests/__init__.py b/xarray/tests/__init__.py
index 0fcfbf98..1ee02abf 100644
--- a/xarray/tests/__init__.py
+++ b/xarray/tests/__init__.py
@@ -104,10 +104,12 @@ class TestCase(unittest.TestCase):
         # Python 3 assertCountEqual is roughly equivalent to Python 2
         # assertItemsEqual
         def assertItemsEqual(self, first, second, msg=None):
+            __tracebackhide__ = True  # noqa: F841
             return self.assertCountEqual(first, second, msg)
 
     @contextmanager
     def assertWarns(self, message):
+        __tracebackhide__ = True  # noqa: F841
         with warnings.catch_warnings(record=True) as w:
             warnings.filterwarnings('always', message)
             yield
@@ -115,50 +117,69 @@ class TestCase(unittest.TestCase):
         assert any(message in str(wi.message) for wi in w)
 
     def assertVariableEqual(self, v1, v2):
+        __tracebackhide__ = True  # noqa: F841
         assert_equal(v1, v2)
 
     def assertVariableIdentical(self, v1, v2):
+        __tracebackhide__ = True  # noqa: F841
         assert_identical(v1, v2)
 
     def assertVariableAllClose(self, v1, v2, rtol=1e-05, atol=1e-08):
+        __tracebackhide__ = True  # noqa: F841
         assert_allclose(v1, v2, rtol=rtol, atol=atol)
 
     def assertVariableNotEqual(self, v1, v2):
+        __tracebackhide__ = True  # noqa: F841
         assert not v1.equals(v2)
 
     def assertArrayEqual(self, a1, a2):
+        __tracebackhide__ = True  # noqa: F841
         assert_array_equal(a1, a2)
 
     def assertEqual(self, a1, a2):
+        __tracebackhide__ = True  # noqa: F841
         assert a1 == a2 or (a1 != a1 and a2 != a2)
 
     def assertAllClose(self, a1, a2, rtol=1e-05, atol=1e-8):
+        __tracebackhide__ = True  # noqa: F841
         assert allclose_or_equiv(a1, a2, rtol=rtol, atol=atol)
 
     def assertDatasetEqual(self, d1, d2):
+        __tracebackhide__ = True  # noqa: F841
         assert_equal(d1, d2)
 
     def assertDatasetIdentical(self, d1, d2):
+        __tracebackhide__ = True  # noqa: F841
         assert_identical(d1, d2)
 
-    def assertDatasetAllClose(self, d1, d2, rtol=1e-05, atol=1e-08):
-        assert_allclose(d1, d2, rtol=rtol, atol=atol)
+    def assertDatasetAllClose(self, d1, d2, rtol=1e-05, atol=1e-08,
+                              decode_bytes=True):
+        __tracebackhide__ = True  # noqa: F841
+        assert_allclose(d1, d2, rtol=rtol, atol=atol,
+                        decode_bytes=decode_bytes)
 
     def assertCoordinatesEqual(self, d1, d2):
+        __tracebackhide__ = True  # noqa: F841
         assert_equal(d1, d2)
 
     def assertDataArrayEqual(self, ar1, ar2):
+        __tracebackhide__ = True  # noqa: F841
         assert_equal(ar1, ar2)
 
     def assertDataArrayIdentical(self, ar1, ar2):
+        __tracebackhide__ = True  # noqa: F841
         assert_identical(ar1, ar2)
 
-    def assertDataArrayAllClose(self, ar1, ar2, rtol=1e-05, atol=1e-08):
-        assert_allclose(ar1, ar2, rtol=rtol, atol=atol)
+    def assertDataArrayAllClose(self, ar1, ar2, rtol=1e-05, atol=1e-08,
+                                decode_bytes=True):
+        __tracebackhide__ = True  # noqa: F841
+        assert_allclose(ar1, ar2, rtol=rtol, atol=atol,
+                        decode_bytes=decode_bytes)
 
 
 @contextmanager
 def raises_regex(error, pattern):
+    __tracebackhide__ = True  # noqa: F841
     with pytest.raises(error) as excinfo:
         yield
     message = str(excinfo.value)
diff --git a/xarray/tests/test_backends.py b/xarray/tests/test_backends.py
index 89b23c45..8cef82ef 100644
--- a/xarray/tests/test_backends.py
+++ b/xarray/tests/test_backends.py
@@ -122,7 +122,7 @@ class TestCommon(TestCase):
         self.assertEqual(actual, 0)
 
 
-class Only32BitTypes(object):
+class NetCDF3Only(object):
     pass
 
 
@@ -168,9 +168,10 @@ class DatasetIOTestCases(object):
     def test_zero_dimensional_variable(self):
         expected = create_test_data()
         expected['float_var'] = ([], 1.0e9, {'units': 'units of awesome'})
-        expected['string_var'] = ([], np.array('foobar', dtype='S'))
+        expected['bytes_var'] = ([], b'foobar')
+        expected['string_var'] = ([], u'foobar')
         with self.roundtrip(expected) as actual:
-            self.assertDatasetAllClose(expected, actual)
+            self.assertDatasetIdentical(expected, actual)
 
     def test_write_store(self):
         expected = create_test_data()
@@ -184,8 +185,7 @@ class DatasetIOTestCases(object):
     def check_dtypes_roundtripped(self, expected, actual):
         for k in expected:
             expected_dtype = expected.variables[k].dtype
-            if (isinstance(self, Only32BitTypes) and
-                    expected_dtype == 'int64'):
+            if (isinstance(self, NetCDF3Only) and expected_dtype == 'int64'):
                 # downcast
                 expected_dtype = np.dtype('int32')
             actual_dtype = actual.variables[k].dtype
@@ -199,7 +199,7 @@ class DatasetIOTestCases(object):
         expected = create_test_data()
         with self.roundtrip(expected) as actual:
             self.check_dtypes_roundtripped(expected, actual)
-            self.assertDatasetAllClose(expected, actual)
+            self.assertDatasetIdentical(expected, actual)
 
     def test_load(self):
         expected = create_test_data()
@@ -216,7 +216,7 @@ class DatasetIOTestCases(object):
                 for k, v in actual.variables.items():
                     if k in vars:
                         self.assertTrue(v._in_memory)
-                self.assertDatasetAllClose(expected, actual)
+                self.assertDatasetIdentical(expected, actual)
 
         with self.assertRaises(AssertionError):
             # make sure the contextmanager works!
@@ -232,7 +232,7 @@ class DatasetIOTestCases(object):
         # verify we can read data even after closing the file
         with self.roundtrip(expected) as ds:
             actual = ds.load()
-        self.assertDatasetAllClose(expected, actual)
+        self.assertDatasetIdentical(expected, actual)
 
     def test_dataset_compute(self):
         expected = create_test_data()
@@ -250,8 +250,8 @@ class DatasetIOTestCases(object):
             for v in computed.variables.values():
                 self.assertTrue(v._in_memory)
 
-            self.assertDatasetAllClose(expected, actual)
-            self.assertDatasetAllClose(expected, computed)
+            self.assertDatasetIdentical(expected, actual)
+            self.assertDatasetIdentical(expected, computed)
 
     def test_pickle(self):
         expected = Dataset({'foo': ('x', [42])})
@@ -289,44 +289,56 @@ class DatasetIOTestCases(object):
     def test_roundtrip_None_variable(self):
         expected = Dataset({None: (('x', 'y'), [[0, 1], [2, 3]])})
         with self.roundtrip(expected) as actual:
-            self.assertDatasetAllClose(expected, actual)
+            self.assertDatasetIdentical(expected, actual)
 
     def test_roundtrip_object_dtype(self):
         floats = np.array([0.0, 0.0, 1.0, 2.0, 3.0], dtype=object)
         floats_nans = np.array([np.nan, np.nan, 1.0, 2.0, 3.0], dtype=object)
-        letters = np.array(['ab', 'cdef', 'g'], dtype=object)
-        letters_nans = np.array(['ab', 'cdef', np.nan], dtype=object)
+        bytes_ = np.array([b'ab', b'cdef', b'g'], dtype=object)
+        bytes_nans = np.array([b'ab', b'cdef', np.nan], dtype=object)
+        strings = np.array([u'ab', u'cdef', u'g'], dtype=object)
+        strings_nans = np.array([u'ab', u'cdef', np.nan], dtype=object)
         all_nans = np.array([np.nan, np.nan], dtype=object)
         original = Dataset({'floats': ('a', floats),
                             'floats_nans': ('a', floats_nans),
-                            'letters': ('b', letters),
-                            'letters_nans': ('b', letters_nans),
+                            'bytes': ('b', bytes_),
+                            'bytes_nans': ('b', bytes_nans),
+                            'strings': ('b', strings),
+                            'strings_nans': ('b', strings_nans),
                             'all_nans': ('c', all_nans),
                             'nan': ([], np.nan)})
         expected = original.copy(deep=True)
-        if isinstance(self, Only32BitTypes):
-            # for netCDF3 tests, expect the results to come back as characters
-            expected['letters_nans'] = expected['letters_nans'].astype('S')
-            expected['letters'] = expected['letters'].astype('S')
         with self.roundtrip(original) as actual:
             try:
                 self.assertDatasetIdentical(expected, actual)
             except AssertionError:
-                # Most stores use '' for nans in strings, but some don't
-                # first try the ideal case (where the store returns exactly)
+                # Most stores use '' for nans in strings, but some don't.
+                # First try the ideal case (where the store returns exactly)
                 # the original Dataset), then try a more realistic case.
-                # ScipyDataTest, NetCDF3ViaNetCDF4DataTest and NetCDF4DataTest
-                # all end up using this case.
-                expected['letters_nans'][-1] = ''
+                # This currently includes all netCDF files when encoding is not
+                # explicitly set.
+                # https://github.com/pydata/xarray/issues/1647
+                expected['bytes_nans'][-1] = b''
+                expected['strings_nans'][-1] = u''
                 self.assertDatasetIdentical(expected, actual)
 
     def test_roundtrip_string_data(self):
         expected = Dataset({'x': ('t', ['ab', 'cdef'])})
         with self.roundtrip(expected) as actual:
-            if isinstance(self, Only32BitTypes):
-                expected['x'] = expected['x'].astype('S')
             self.assertDatasetIdentical(expected, actual)
 
+    def test_roundtrip_string_encoded_characters(self):
+        expected = Dataset({'x': ('t', [u'ab', u'cdef'])})
+        expected['x'].encoding['dtype'] = 'S1'
+        with self.roundtrip(expected) as actual:
+            self.assertDatasetIdentical(expected, actual)
+            self.assertEqual(actual['x'].encoding['_Encoding'], 'utf-8')
+
+        expected['x'].encoding['_Encoding'] = 'ascii'
+        with self.roundtrip(expected) as actual:
+            self.assertDatasetIdentical(expected, actual)
+            self.assertEqual(actual['x'].encoding['_Encoding'], 'ascii')
+
     def test_roundtrip_datetime_data(self):
         times = pd.to_datetime(['2000-01-01', '2000-01-02', 'NaT'])
         expected = Dataset({'t': ('t', times), 't0': times[0]})
@@ -382,43 +394,38 @@ class DatasetIOTestCases(object):
                         'dim3': np.arange(5)}
             expected = in_memory.isel(**indexers)
             actual = on_disk.isel(**indexers)
-            self.assertDatasetAllClose(expected, actual)
+            self.assertDatasetIdentical(expected, actual)
             # do it twice, to make sure we're switched from orthogonal -> numpy
             # when we cached the values
             actual = on_disk.isel(**indexers)
-            self.assertDatasetAllClose(expected, actual)
+            self.assertDatasetIdentical(expected, actual)
 
 
 class CFEncodedDataTest(DatasetIOTestCases):
 
-    def test_roundtrip_strings_with_fill_value(self):
-        values = np.array(['ab', 'cdef', np.nan], dtype=object)
-        encoding = {'_FillValue': np.string_('X'), 'dtype': np.dtype('S1')}
+    def test_roundtrip_bytes_with_fill_value(self):
+        values = np.array([b'ab', b'cdef', np.nan], dtype=object)
+        encoding = {'_FillValue': b'X', 'dtype': 'S1'}
         original = Dataset({'x': ('t', values, {}, encoding)})
         expected = original.copy(deep=True)
-        expected['x'][:2] = values[:2].astype('S')
         with self.roundtrip(original) as actual:
             self.assertDatasetIdentical(expected, actual)
 
-        original = Dataset({'x': ('t', values, {}, {'_FillValue': '\x00'})})
-        if not isinstance(self, Only32BitTypes):
-            # these stores can save unicode strings
-            expected = original.copy(deep=True)
-        if isinstance(self, BaseNetCDF4Test):
-            # netCDF4 can't keep track of an empty _FillValue for VLEN
-            # variables
-            expected['x'][-1] = ''
-        elif (isinstance(self, (NetCDF3ViaNetCDF4DataTest,
-                                NetCDF4ClassicViaNetCDF4DataTest))
-              or (has_netCDF4 and
-                  (type(self) is GenericNetCDFDataTest or
-                   type(self) is GenericNetCDFDataTestAutocloseTrue))):
-            # netCDF4 can't keep track of an empty _FillValue for nc3, either:
-            # https://github.com/Unidata/netcdf4-python/issues/273
-            expected['x'][-1] = np.string_('')
+        original = Dataset({'x': ('t', values, {}, {'_FillValue': b''})})
         with self.roundtrip(original) as actual:
             self.assertDatasetIdentical(expected, actual)
 
+    def test_roundtrip_string_with_fill_value_nchar(self):
+        values = np.array([u'ab', u'cdef', np.nan], dtype=object)
+        expected = Dataset({'x': ('t', values)})
+
+        encoding = {'dtype': 'S1', '_FillValue': b'X'}
+        original = Dataset({'x': ('t', values, {}, encoding)})
+        # Not supported yet.
+        with pytest.raises(NotImplementedError):
+            with self.roundtrip(original) as actual:
+                self.assertDatasetIdentical(expected, actual)
+
     def test_unsigned_roundtrip_mask_and_scale(self):
         decoded = create_unsigned_masked_scaled_data()
         encoded = create_encoded_unsigned_masked_scaled_data()
@@ -426,57 +433,58 @@ class CFEncodedDataTest(DatasetIOTestCases):
             for k in decoded.variables:
                 self.assertEqual(decoded.variables[k].dtype,
                                  actual.variables[k].dtype)
-            self.assertDatasetAllClose(decoded, actual)
+            self.assertDatasetAllClose(decoded, actual, decode_bytes=False)
         with self.roundtrip(decoded,
                             open_kwargs=dict(decode_cf=False)) as actual:
             for k in encoded.variables:
                 self.assertEqual(encoded.variables[k].dtype,
                                  actual.variables[k].dtype)
-            self.assertDatasetAllClose(encoded, actual)
+            self.assertDatasetAllClose(encoded, actual, decode_bytes=False)
         with self.roundtrip(encoded,
                             open_kwargs=dict(decode_cf=False)) as actual:
             for k in encoded.variables:
                 self.assertEqual(encoded.variables[k].dtype,
                                  actual.variables[k].dtype)
-            self.assertDatasetAllClose(encoded, actual)
+            self.assertDatasetAllClose(encoded, actual, decode_bytes=False)
         # make sure roundtrip encoding didn't change the
         # original dataset.
-        self.assertDatasetIdentical(
+        self.assertDatasetAllClose(
             encoded, create_encoded_unsigned_masked_scaled_data())
         with self.roundtrip(encoded) as actual:
             for k in decoded.variables:
                 self.assertEqual(decoded.variables[k].dtype,
                                  actual.variables[k].dtype)
-            self.assertDatasetAllClose(decoded, actual)
+            self.assertDatasetAllClose(decoded, actual, decode_bytes=False)
         with self.roundtrip(encoded,
                             open_kwargs=dict(decode_cf=False)) as actual:
             for k in encoded.variables:
                 self.assertEqual(encoded.variables[k].dtype,
                                  actual.variables[k].dtype)
-            self.assertDatasetAllClose(encoded, actual)
+            self.assertDatasetAllClose(encoded, actual, decode_bytes=False)
 
     def test_roundtrip_mask_and_scale(self):
         decoded = create_masked_and_scaled_data()
         encoded = create_encoded_masked_and_scaled_data()
         with self.roundtrip(decoded) as actual:
-            self.assertDatasetAllClose(decoded, actual)
+            self.assertDatasetAllClose(decoded, actual, decode_bytes=False)
         with self.roundtrip(decoded,
                             open_kwargs=dict(decode_cf=False)) as actual:
             # TODO: this assumes that all roundtrips will first
             # encode.  Is that something we want to test for?
-            self.assertDatasetAllClose(encoded, actual)
+            self.assertDatasetAllClose(encoded, actual, decode_bytes=False)
         with self.roundtrip(encoded,
                             open_kwargs=dict(decode_cf=False)) as actual:
-            self.assertDatasetAllClose(encoded, actual)
+            self.assertDatasetAllClose(encoded, actual, decode_bytes=False)
         # make sure roundtrip encoding didn't change the
         # original dataset.
-        self.assertDatasetIdentical(encoded,
-                                    create_encoded_masked_and_scaled_data())
+        self.assertDatasetAllClose(encoded,
+                                   create_encoded_masked_and_scaled_data(),
+                                   decode_bytes=False)
         with self.roundtrip(encoded) as actual:
-            self.assertDatasetAllClose(decoded, actual)
+            self.assertDatasetAllClose(decoded, actual, decode_bytes=False)
         with self.roundtrip(encoded,
                             open_kwargs=dict(decode_cf=False)) as actual:
-            self.assertDatasetAllClose(encoded, actual)
+            self.assertDatasetAllClose(encoded, actual, decode_bytes=False)
 
     def test_coordinates_encoding(self):
         def equals_latlon(obj):
@@ -603,7 +611,7 @@ class CFEncodedDataTest(DatasetIOTestCases):
         # regression for GH1215
         data = create_test_data()
         with self.roundtrip_append(data) as actual:
-            assert_allclose(data, actual)
+            self.assertDatasetIdentical(data, actual)
 
     def test_append_overwrite_values(self):
         # regression for GH1215
@@ -614,7 +622,7 @@ class CFEncodedDataTest(DatasetIOTestCases):
             data['var9'] = data['var2'] * 3
             self.save(data[['var2', 'var9']], tmp_file, mode='a')
             with self.open(tmp_file) as actual:
-                assert_allclose(data, actual)
+                self.assertDatasetIdentical(data, actual)
 
 
 _counter = itertools.count()
@@ -705,6 +713,24 @@ class BaseNetCDF4Test(CFEncodedDataTest):
             with open_dataset(tmp_file, group='data/2') as actual2:
                 self.assertDatasetIdentical(data2, actual2)
 
+    def test_roundtrip_string_with_fill_value_vlen(self):
+        values = np.array([u'ab', u'cdef', np.nan], dtype=object)
+        expected = Dataset({'x': ('t', values)})
+
+        # netCDF4-based backends don't support an explicit fillvalue
+        # for variable length strings yet.
+        # https://github.com/Unidata/netcdf4-python/issues/730
+        # https://github.com/shoyer/h5netcdf/issues/37
+        original = Dataset({'x': ('t', values, {}, {'_FillValue': u'XXX'})})
+        with pytest.raises(NotImplementedError):
+            with self.roundtrip(original) as actual:
+                self.assertDatasetIdentical(expected, actual)
+
+        original = Dataset({'x': ('t', values, {}, {'_FillValue': u''})})
+        with pytest.raises(NotImplementedError):
+            with self.roundtrip(original) as actual:
+                self.assertDatasetIdentical(expected, actual)
+
     def test_roundtrip_character_array(self):
         with create_tmp_file() as tmp_file:
             values = np.array([['a', 'b', 'c'], ['d', 'e', 'f']], dtype='S')
@@ -928,7 +954,7 @@ class NetCDF4ViaDaskDataTestAutocloseTrue(NetCDF4ViaDaskDataTest):
 
 
 @requires_scipy
-class ScipyInMemoryDataTest(CFEncodedDataTest, Only32BitTypes, TestCase):
+class ScipyInMemoryDataTest(CFEncodedDataTest, NetCDF3Only, TestCase):
     engine = 'scipy'
 
     @contextlib.contextmanager
@@ -954,7 +980,7 @@ class ScipyInMemoryDataTestAutocloseTrue(ScipyInMemoryDataTest):
 
 
 @requires_scipy
-class ScipyFileObjectTest(CFEncodedDataTest, Only32BitTypes, TestCase):
+class ScipyFileObjectTest(CFEncodedDataTest, NetCDF3Only, TestCase):
     engine = 'scipy'
 
     @contextlib.contextmanager
@@ -982,7 +1008,7 @@ class ScipyFileObjectTest(CFEncodedDataTest, Only32BitTypes, TestCase):
 
 
 @requires_scipy
-class ScipyFilePathTest(CFEncodedDataTest, Only32BitTypes, TestCase):
+class ScipyFilePathTest(CFEncodedDataTest, NetCDF3Only, TestCase):
     engine = 'scipy'
 
     @contextlib.contextmanager
@@ -998,14 +1024,9 @@ class ScipyFilePathTest(CFEncodedDataTest, Only32BitTypes, TestCase):
                 pass
 
     def test_roundtrip_example_1_netcdf_gz(self):
-        if sys.version_info[:2] < (2, 7):
-            with self.assertRaisesRegexp(ValueError,
-                                         'gzipped netCDF not supported'):
-                open_example_dataset('example_1.nc.gz')
-        else:
-            with open_example_dataset('example_1.nc.gz') as expected:
-                with open_example_dataset('example_1.nc') as actual:
-                    self.assertDatasetIdentical(expected, actual)
+        with open_example_dataset('example_1.nc.gz') as expected:
+            with open_example_dataset('example_1.nc') as actual:
+                self.assertDatasetIdentical(expected, actual)
 
     def test_netcdf3_endianness(self):
         # regression test for GH416
@@ -1028,7 +1049,7 @@ class ScipyFilePathTestAutocloseTrue(ScipyFilePathTest):
 
 
 @requires_netCDF4
-class NetCDF3ViaNetCDF4DataTest(CFEncodedDataTest, Only32BitTypes, TestCase):
+class NetCDF3ViaNetCDF4DataTest(CFEncodedDataTest, NetCDF3Only, TestCase):
     engine = 'netcdf4'
     file_format = 'NETCDF3_CLASSIC'
 
@@ -1045,7 +1066,7 @@ class NetCDF3ViaNetCDF4DataTestAutocloseTrue(NetCDF3ViaNetCDF4DataTest):
 
 
 @requires_netCDF4
-class NetCDF4ClassicViaNetCDF4DataTest(CFEncodedDataTest, Only32BitTypes,
+class NetCDF4ClassicViaNetCDF4DataTest(CFEncodedDataTest, NetCDF3Only,
                                        TestCase):
     engine = 'netcdf4'
     file_format = 'NETCDF4_CLASSIC'
@@ -1064,7 +1085,7 @@ class NetCDF4ClassicViaNetCDF4DataTestAutocloseTrue(
 
 
 @requires_scipy_or_netCDF4
-class GenericNetCDFDataTest(CFEncodedDataTest, Only32BitTypes, TestCase):
+class GenericNetCDFDataTest(CFEncodedDataTest, NetCDF3Only, TestCase):
     # verify that we can read and write netCDF3 files as long as we have scipy
     # or netCDF4-python installed
     file_format = 'netcdf3_64bit'
@@ -1390,8 +1411,13 @@ class DaskTest(TestCase, DatasetIOTestCases):
                   allow_cleanup_failure=False):
         yield data.chunk()
 
+    def test_roundtrip_string_encoded_characters(self):
+        # Override method in DatasetIOTestCases - not applicable to dask
+        pass
+
     def test_roundtrip_datetime_data(self):
-        # Override method in DatasetIOTestCases - remove not applicable save_kwds
+        # Override method in DatasetIOTestCases - remove not applicable
+        # save_kwds
         times = pd.to_datetime(['2000-01-01', '2000-01-02', 'NaT'])
         expected = Dataset({'t': ('t', times), 't0': times[0]})
         with self.roundtrip(expected) as actual:
@@ -1419,7 +1445,7 @@ class DaskTest(TestCase, DatasetIOTestCases):
                     self.assertIsInstance(actual.foo.variable.data, da.Array)
                     self.assertEqual(actual.foo.variable.data.chunks,
                                      ((5, 5),))
-                    self.assertDatasetAllClose(original, actual)
+                    self.assertDatasetIdentical(original, actual)
                 with open_mfdataset([tmp1, tmp2], chunks={'x': 3},
                                     autoclose=self.autoclose) as actual:
                     self.assertEqual(actual.foo.variable.data.chunks,
@@ -1439,7 +1465,7 @@ class DaskTest(TestCase, DatasetIOTestCases):
                 original.isel(x=slice(5, 10)).to_netcdf(tmp2)
                 with open_mfdataset([tmp1, tmp2],
                                     autoclose=self.autoclose) as actual:
-                    self.assertDatasetAllClose(original, actual)
+                    self.assertDatasetIdentical(original, actual)
 
     def test_attrs_mfdataset(self):
         original = Dataset({'foo': ('x', np.random.randn(10))})
@@ -1518,7 +1544,8 @@ class DaskTest(TestCase, DatasetIOTestCases):
             original.to_netcdf(tmp)
             with open_mfdataset(tmp, autoclose=self.autoclose) as ds:
                 actual = 1.0 * ds
-                self.assertDatasetAllClose(original, actual)
+                self.assertDatasetAllClose(original, actual,
+                                           decode_bytes=False)
 
     def test_open_mfdataset_concat_dim_none(self):
         with create_tmp_file() as tmp1:
@@ -1579,7 +1606,7 @@ class DaskTest(TestCase, DatasetIOTestCases):
         computed = actual.compute()
         self.assertFalse(actual._in_memory)
         self.assertTrue(computed._in_memory)
-        self.assertDataArrayAllClose(actual, computed)
+        self.assertDataArrayAllClose(actual, computed, decode_bytes=False)
 
 
 class DaskTestAutocloseTrue(DaskTest):
@@ -1637,7 +1664,7 @@ class PydapTest(TestCase):
 
 @requires_scipy
 @requires_pynio
-class TestPyNio(CFEncodedDataTest, Only32BitTypes, TestCase):
+class TestPyNio(CFEncodedDataTest, NetCDF3Only, TestCase):
     def test_write_store(self):
         # pynio is read-only for now
         pass
diff --git a/xarray/tests/test_conventions.py b/xarray/tests/test_conventions.py
index 6aaa2cbf..d5dfa289 100644
--- a/xarray/tests/test_conventions.py
+++ b/xarray/tests/test_conventions.py
@@ -1,9 +1,11 @@
+# -*- coding: utf-8 -*-
 from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
 import contextlib
 import numpy as np
 import pandas as pd
+import pytest
 import warnings
 
 from xarray import conventions, Variable, Dataset, open_dataset
@@ -56,25 +58,11 @@ class TestMaskedAndScaledArray(TestCase):
         self.assertTrue(np.isnan(x[...]))
 
 
-class TestCharToStringArray(TestCase):
+class TestStackedBytesArray(TestCase):
     def test_wrapper_class(self):
-        array = np.array(list('abc'), dtype='S')
-        actual = conventions.CharToStringArray(array)
-        expected = np.array('abc', dtype='S')
-        self.assertEqual(actual.dtype, expected.dtype)
-        self.assertEqual(actual.shape, expected.shape)
-        self.assertEqual(actual.size, expected.size)
-        self.assertEqual(actual.ndim, expected.ndim)
-        with self.assertRaises(TypeError):
-            len(actual)
-        self.assertArrayEqual(expected, actual)
-        with self.assertRaises(IndexError):
-            actual[:2]
-        self.assertEqual(str(actual), 'abc')
-
-        array = np.array([list('abc'), list('cdf')], dtype='S')
-        actual = conventions.CharToStringArray(array)
-        expected = np.array(['abc', 'cdf'], dtype='S')
+        array = np.array([[b'a', b'b', b'c'], [b'd', b'e', b'f']], dtype='S')
+        actual = conventions.StackedBytesArray(array)
+        expected = np.array([b'abc', b'def'], dtype='S')
         self.assertEqual(actual.dtype, expected.dtype)
         self.assertEqual(actual.shape, expected.shape)
         self.assertEqual(actual.size, expected.size)
@@ -85,29 +73,94 @@ class TestCharToStringArray(TestCase):
         with self.assertRaises(IndexError):
             actual[:, :2]
 
-    def test_char_to_string(self):
+    def test_scalar(self):
+        array = np.array([b'a', b'b', b'c'], dtype='S')
+        actual = conventions.StackedBytesArray(array)
+
+        expected = np.array(b'abc')
+        assert actual.dtype == expected.dtype
+        assert actual.shape == expected.shape
+        assert actual.size == expected.size
+        assert actual.ndim == expected.ndim
+        with pytest.raises(TypeError):
+            len(actual)
+        np.testing.assert_array_equal(expected, actual)
+        with pytest.raises(IndexError):
+            actual[:2]
+        assert str(actual) == str(expected)
+
+    def test_char_to_bytes(self):
         array = np.array([['a', 'b', 'c'], ['d', 'e', 'f']])
         expected = np.array(['abc', 'def'])
-        actual = conventions.char_to_string(array)
+        actual = conventions.char_to_bytes(array)
         self.assertArrayEqual(actual, expected)
 
         expected = np.array(['ad', 'be', 'cf'])
-        actual = conventions.char_to_string(array.T)  # non-contiguous
+        actual = conventions.char_to_bytes(array.T)  # non-contiguous
+        self.assertArrayEqual(actual, expected)
+
+    def test_char_to_bytes_ndim_zero(self):
+        expected = np.array('a')
+        actual = conventions.char_to_bytes(expected)
         self.assertArrayEqual(actual, expected)
 
-    def test_string_to_char(self):
+    def test_char_to_bytes_size_zero(self):
+        array = np.zeros((3, 0), dtype='S1')
+        expected = np.array([b'', b'', b''])
+        actual = conventions.char_to_bytes(array)
+        self.assertArrayEqual(actual, expected)
+
+    def test_bytes_to_char(self):
         array = np.array([['ab', 'cd'], ['ef', 'gh']])
         expected = np.array([[['a', 'b'], ['c', 'd']],
                              [['e', 'f'], ['g', 'h']]])
-        actual = conventions.string_to_char(array)
+        actual = conventions.bytes_to_char(array)
         self.assertArrayEqual(actual, expected)
 
         expected = np.array([[['a', 'b'], ['e', 'f']],
                              [['c', 'd'], ['g', 'h']]])
-        actual = conventions.string_to_char(array.T)
+        actual = conventions.bytes_to_char(array.T)
         self.assertArrayEqual(actual, expected)
 
 
+class TestBytesToStringArray(TestCase):
+
+    def test_encoding(self):
+        encoding = 'utf-8'
+        raw_array = np.array([b'abc', u'ß∂µ∆'.encode(encoding)])
+        actual = conventions.BytesToStringArray(raw_array, encoding=encoding)
+        expected = np.array([u'abc', u'ß∂µ∆'], dtype=object)
+
+        self.assertEqual(actual.dtype, expected.dtype)
+        self.assertEqual(actual.shape, expected.shape)
+        self.assertEqual(actual.size, expected.size)
+        self.assertEqual(actual.ndim, expected.ndim)
+        self.assertArrayEqual(expected, actual)
+        self.assertArrayEqual(expected[0], actual[0])
+
+    def test_scalar(self):
+        expected = np.array(u'abc', dtype=object)
+        actual = conventions.BytesToStringArray(
+            np.array(b'abc'), encoding='utf-8')
+        assert actual.dtype == expected.dtype
+        assert actual.shape == expected.shape
+        assert actual.size == expected.size
+        assert actual.ndim == expected.ndim
+        with pytest.raises(TypeError):
+            len(actual)
+        np.testing.assert_array_equal(expected, actual)
+        with pytest.raises(IndexError):
+            actual[:2]
+        assert str(actual) == str(expected)
+
+    def test_decode_bytes_array(self):
+        encoding = 'utf-8'
+        raw_array = np.array([b'abc', u'ß∂µ∆'.encode(encoding)])
+        expected = np.array([u'abc', u'ß∂µ∆'], dtype=object)
+        actual = conventions.decode_bytes_array(raw_array, encoding)
+        np.testing.assert_array_equal(actual, expected)
+
+
 class TestUnsignedIntTypeArray(TestCase):
     def test_unsignedinttype_array(self):
         sb = np.asarray([0, 1, 127, -128, -1], dtype='i1')
@@ -268,21 +321,22 @@ class TestDatetime(TestCase):
                        {'units': 'foobar',
                         'missing_value': 0,
                         '_FillValue': 1})
-        self.assertRaisesRegexp(ValueError, "_FillValue and missing_value",
-                                lambda: conventions.decode_cf_variable(var))
+        with self.assertRaisesRegexp(
+                ValueError, "_FillValue and missing_value"):
+            conventions.decode_cf_variable('t', var)
 
         var = Variable(['t'], np.arange(10),
                        {'units': 'foobar',
                         'missing_value': np.nan,
                         '_FillValue': np.nan})
-        var = conventions.decode_cf_variable(var)
+        var = conventions.decode_cf_variable('t', var)
         self.assertIsNotNone(var)
 
         var = Variable(['t'], np.arange(10),
                                {'units': 'foobar',
                                 'missing_value': np.float32(np.nan),
                                 '_FillValue': np.float32(np.nan)})
-        var = conventions.decode_cf_variable(var)
+        var = conventions.decode_cf_variable('t', var)
         self.assertIsNotNone(var)
 
     @requires_netCDF4
@@ -578,7 +632,7 @@ class TestDecodeCF(TestCase):
     def test_0d_int32_encoding(self):
         original = Variable((), np.int32(0), encoding={'dtype': 'int64'})
         expected = Variable((), np.int64(0))
-        actual = conventions.maybe_encode_dtype(original)
+        actual = conventions.maybe_encode_nonstring_dtype(original)
         self.assertDatasetIdentical(expected, actual)
 
     def test_decode_cf_with_multiple_missing_values(self):
@@ -586,9 +640,9 @@ class TestDecodeCF(TestCase):
                             {'missing_value': np.array([0, 1])})
         expected = Variable(['t'], [np.nan, np.nan, 2], {})
         with warnings.catch_warnings(record=True) as w:
-            actual = conventions.decode_cf_variable(original)
+            actual = conventions.decode_cf_variable('t', original)
             self.assertDatasetIdentical(expected, actual)
-            self.assertIn('variable has multiple fill', str(w[0].message))
+            self.assertIn('has multiple fill', str(w[0].message))
 
     def test_decode_cf_with_drop_variables(self):
         original = Dataset({