diff --git a/gcloud/datastore/__init__.py b/gcloud/datastore/__init__.py
index 1a2df8147706..e717a4ac5052 100644
--- a/gcloud/datastore/__init__.py
+++ b/gcloud/datastore/__init__.py
@@ -69,6 +69,48 @@
 _DATASET_ENV_VAR_NAME = 'GCLOUD_DATASET_ID'
 
 
+def _find_true_dataset_id(dataset_id, connection=None):
+    """Find the true (unaliased) dataset ID.
+
+    If the given ID already has a 's~' or 'e~' prefix, does nothing.
+    Otherwise, looks up a bogus Key('__MissingLookupKind', 1) and reads the
+    true prefixed dataset ID from the response (either from found or from
+    missing).
+
+    For some context, see:
+      github.com/GoogleCloudPlatform/gcloud-python/pull/528
+      github.com/GoogleCloudPlatform/google-cloud-datastore/issues/59
+
+    :type dataset_id: string
+    :param dataset_id: The dataset ID to un-alias / prefix.
+
+    :type connection: :class:`gcloud.datastore.connection.Connection`
+    :param connection: Optional. A connection provided to be the default.
+
+    :rtype: string
+    :returns: The true / prefixed / un-aliased dataset ID.
+    """
+    if dataset_id.startswith('s~') or dataset_id.startswith('e~'):
+        return dataset_id
+
+    connection = connection or _implicit_environ.CONNECTION
+
+    # Create the bogus Key protobuf to be looked up and remove
+    # the dataset ID so the backend won't complain.
+    bogus_key_pb = Key('__MissingLookupKind', 1,
+                       dataset_id=dataset_id).to_protobuf()
+    bogus_key_pb.partition_id.ClearField('dataset_id')
+
+    found_pbs, missing_pbs, _ = connection.lookup(dataset_id, [bogus_key_pb])
+    # By not passing in `deferred`, lookup will continue until
+    # all results are `found` or `missing`.
+    all_pbs = missing_pbs + found_pbs
+    # We only asked for one, so should only receive one.
+    returned_pb, = all_pbs
+
+    return returned_pb.key.partition_id.dataset_id
+
+
 def set_default_dataset_id(dataset_id=None):
     """Set default dataset ID either explicitly or implicitly as fall-back.
 
@@ -91,6 +133,7 @@ def set_default_dataset_id(dataset_id=None):
         dataset_id = _implicit_environ.compute_engine_id()
 
     if dataset_id is not None:
+        dataset_id = _find_true_dataset_id(dataset_id)
         _implicit_environ.DATASET_ID = dataset_id
 
 
@@ -120,8 +163,9 @@ def set_defaults(dataset_id=None, connection=None):
     :type connection: :class:`gcloud.datastore.connection.Connection`
     :param connection: A connection provided to be the default.
     """
-    set_default_dataset_id(dataset_id=dataset_id)
+    # Set CONNECTION first in case _find_true_dataset_id needs it.
     set_default_connection(connection=connection)
+    set_default_dataset_id(dataset_id=dataset_id)
 
 
 def get_connection():
diff --git a/gcloud/datastore/batch.py b/gcloud/datastore/batch.py
index bc9c370fdeea..3b5b47f6cfa7 100644
--- a/gcloud/datastore/batch.py
+++ b/gcloud/datastore/batch.py
@@ -216,8 +216,7 @@ def put(self, entity):
         if entity.key is None:
             raise ValueError("Entity must have a key")
 
-        if not helpers._dataset_ids_equal(self._dataset_id,
-                                          entity.key.dataset_id):
+        if self._dataset_id != entity.key.dataset_id:
             raise ValueError("Key must be from same dataset as batch")
 
         _assign_entity_to_mutation(
@@ -235,8 +234,7 @@ def delete(self, key):
         if key.is_partial:
             raise ValueError("Key must be complete")
 
-        if not helpers._dataset_ids_equal(self._dataset_id,
-                                          key.dataset_id):
+        if self._dataset_id != key.dataset_id:
             raise ValueError("Key must be from same dataset as batch")
 
         key_pb = key.to_protobuf()
@@ -309,7 +307,6 @@ def _assign_entity_to_mutation(mutation_pb, entity, auto_id_entities):
     auto_id = entity.key.is_partial
 
     key_pb = entity.key.to_protobuf()
-    key_pb = helpers._prepare_key_for_request(key_pb)
 
     if auto_id:
         insert = mutation_pb.insert_auto_id.add()
diff --git a/gcloud/datastore/helpers.py b/gcloud/datastore/helpers.py
index c6178b369ad3..c29ff7e632f6 100644
--- a/gcloud/datastore/helpers.py
+++ b/gcloud/datastore/helpers.py
@@ -26,7 +26,6 @@
 import pytz
 import six
 
-from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
 from gcloud.datastore.entity import Entity
 from gcloud.datastore.key import Key
 
@@ -280,33 +279,6 @@ def _set_protobuf_value(value_pb, val):
         setattr(value_pb, attr, val)
 
 
-def _prepare_key_for_request(key_pb):
-    """Add protobuf keys to a request object.
-
-    :type key_pb: :class:`gcloud.datastore._datastore_v1_pb2.Key`
-    :param key_pb: A key to be added to a request.
-
-    :rtype: :class:`gcloud.datastore._datastore_v1_pb2.Key`
-    :returns: A key which will be added to a request. It will be the
-              original if nothing needs to be changed.
-    """
-    if key_pb.partition_id.HasField('dataset_id'):
-        # We remove the dataset_id from the protobuf. This is because
-        # the backend fails a request if the key contains un-prefixed
-        # dataset ID. The backend fails because requests to
-        #     /datastore/.../datasets/foo/...
-        # and
-        #     /datastore/.../datasets/s~foo/...
-        # both go to the datastore given by 's~foo'. So if the key
-        # protobuf in the request body has dataset_id='foo', the
-        # backend will reject since 'foo' != 's~foo'.
-        new_key_pb = datastore_pb.Key()
-        new_key_pb.CopyFrom(key_pb)
-        new_key_pb.partition_id.ClearField('dataset_id')
-        key_pb = new_key_pb
-    return key_pb
-
-
 def _add_keys_to_request(request_field_pb, key_pbs):
     """Add protobuf keys to a request object.
 
@@ -317,57 +289,4 @@ def _add_keys_to_request(request_field_pb, key_pbs):
     :param key_pbs: The keys to add to a request.
     """
     for key_pb in key_pbs:
-        key_pb = _prepare_key_for_request(key_pb)
         request_field_pb.add().CopyFrom(key_pb)
-
-
-def _dataset_ids_equal(dataset_id1, dataset_id2):
-    """Compares two dataset IDs for fuzzy equality.
-
-    Each may be prefixed or unprefixed (but not null, since dataset ID
-    is required on a key). The only allowed prefixes are 's~' and 'e~'.
-
-    Two identical prefixed match
-
-      >>> 's~foo' == 's~foo'
-      >>> 'e~bar' == 'e~bar'
-
-    while non-identical prefixed don't
-
-      >>> 's~foo' != 's~bar'
-      >>> 's~foo' != 'e~foo'
-
-    As for non-prefixed, they can match other non-prefixed or
-    prefixed:
-
-      >>> 'foo' == 'foo'
-      >>> 'foo' == 's~foo'
-      >>> 'foo' == 'e~foo'
-      >>> 'foo' != 'bar'
-      >>> 'foo' != 's~bar'
-
-    (Ties are resolved since 'foo' can only be an alias for one of
-    s~foo or e~foo in the backend.)
-
-    :type dataset_id1: string
-    :param dataset_id1: A dataset ID.
-
-    :type dataset_id2: string
-    :param dataset_id2: A dataset ID.
-
-    :rtype: boolean
-    :returns: Boolean indicating if the IDs are the same.
-    """
-    if dataset_id1 == dataset_id2:
-        return True
-
-    if dataset_id1.startswith('s~') or dataset_id1.startswith('e~'):
-        # If `dataset_id1` is prefixed and not matching, then the only way
-        # they can match is if `dataset_id2` is unprefixed.
-        return dataset_id1[2:] == dataset_id2
-    elif dataset_id2.startswith('s~') or dataset_id2.startswith('e~'):
-        # Here we know `dataset_id1` is unprefixed and `dataset_id2`
-        # is prefixed.
-        return dataset_id1 == dataset_id2[2:]
-
-    return False
diff --git a/gcloud/datastore/query.py b/gcloud/datastore/query.py
index 97f86efb1f23..48504e16b7e2 100644
--- a/gcloud/datastore/query.py
+++ b/gcloud/datastore/query.py
@@ -449,8 +449,7 @@ def _pb_from_query(query):
     composite_filter.operator = datastore_pb.CompositeFilter.AND
 
     if query.ancestor:
-        ancestor_pb = helpers._prepare_key_for_request(
-            query.ancestor.to_protobuf())
+        ancestor_pb = query.ancestor.to_protobuf()
 
         # Filter on __key__ HAS_ANCESTOR == ancestor.
         ancestor_filter = composite_filter.filter.add().property_filter
@@ -469,8 +468,7 @@ def _pb_from_query(query):
         # Set the value to filter on based on the type.
         if property_name == '__key__':
             key_pb = value.to_protobuf()
-            property_filter.value.key_value.CopyFrom(
-                helpers._prepare_key_for_request(key_pb))
+            property_filter.value.key_value.CopyFrom(key_pb)
         else:
             helpers._set_protobuf_value(property_filter.value, value)
 
diff --git a/gcloud/datastore/test___init__.py b/gcloud/datastore/test___init__.py
index 0ebc28a8927c..d3790f5832d6 100644
--- a/gcloud/datastore/test___init__.py
+++ b/gcloud/datastore/test___init__.py
@@ -64,7 +64,7 @@ def test_no_env_var_set(self):
 
     def test_set_from_env_var(self):
         from gcloud.datastore import _implicit_environ
-        IMPLICIT_DATASET_ID = 'IMPLICIT'
+        IMPLICIT_DATASET_ID = 's~IMPLICIT'
 
         with self._monkeyEnviron(IMPLICIT_DATASET_ID):
             with self._monkeyImplicit():
@@ -74,7 +74,7 @@ def test_set_from_env_var(self):
 
     def test_set_explicit_w_env_var_set(self):
         from gcloud.datastore import _implicit_environ
-        EXPLICIT_DATASET_ID = 'EXPLICIT'
+        EXPLICIT_DATASET_ID = 's~EXPLICIT'
 
         with self._monkeyEnviron(None):
             with self._monkeyImplicit():
@@ -84,8 +84,8 @@ def test_set_explicit_w_env_var_set(self):
 
     def test_set_explicit_no_env_var_set(self):
         from gcloud.datastore import _implicit_environ
-        IMPLICIT_DATASET_ID = 'IMPLICIT'
-        EXPLICIT_DATASET_ID = 'EXPLICIT'
+        IMPLICIT_DATASET_ID = 's~IMPLICIT'
+        EXPLICIT_DATASET_ID = 's~EXPLICIT'
 
         with self._monkeyEnviron(IMPLICIT_DATASET_ID):
             with self._monkeyImplicit():
@@ -104,7 +104,7 @@ def test_set_explicit_None_wo_env_var_set(self):
 
     def test_set_explicit_None_w_env_var_set(self):
         from gcloud.datastore import _implicit_environ
-        IMPLICIT_DATASET_ID = 'IMPLICIT'
+        IMPLICIT_DATASET_ID = 's~IMPLICIT'
 
         with self._monkeyEnviron(IMPLICIT_DATASET_ID):
             with self._monkeyImplicit():
@@ -115,7 +115,7 @@ def test_set_explicit_None_w_env_var_set(self):
     def test_set_implicit_from_appengine(self):
         from gcloud.datastore import _implicit_environ
 
-        APP_ENGINE_ID = 'GAE'
+        APP_ENGINE_ID = 's~GAE'
         APP_IDENTITY = _AppIdentity(APP_ENGINE_ID)
 
         with self._monkeyEnviron(None):
@@ -127,8 +127,8 @@ def test_set_implicit_from_appengine(self):
     def test_set_implicit_both_env_and_appengine(self):
         from gcloud.datastore import _implicit_environ
 
-        IMPLICIT_DATASET_ID = 'IMPLICIT'
-        APP_IDENTITY = _AppIdentity('GAE')
+        IMPLICIT_DATASET_ID = 's~IMPLICIT'
+        APP_IDENTITY = _AppIdentity('s~GAE')
 
         with self._monkeyEnviron(IMPLICIT_DATASET_ID):
             with self._monkeyImplicit(app_identity=APP_IDENTITY):
@@ -139,7 +139,7 @@ def test_set_implicit_both_env_and_appengine(self):
     def _implicit_compute_engine_helper(self, status):
         from gcloud.datastore import _implicit_environ
 
-        COMPUTE_ENGINE_ID = 'GCE'
+        COMPUTE_ENGINE_ID = 's~GCE'
         if status == 200:
             EXPECTED_ID = COMPUTE_ENGINE_ID
         else:
@@ -180,9 +180,9 @@ def test_set_implicit_from_compute_engine_raise_timeout(self):
     def test_set_implicit_both_appengine_and_compute(self):
         from gcloud.datastore import _implicit_environ
 
-        APP_ENGINE_ID = 'GAE'
+        APP_ENGINE_ID = 's~GAE'
         APP_IDENTITY = _AppIdentity(APP_ENGINE_ID)
-        connection = _HTTPConnection(200, 'GCE')
+        connection = _HTTPConnection(200, 's~GCE')
 
         with self._monkeyEnviron(None):
             with self._monkeyImplicit(connection=connection,
@@ -196,9 +196,9 @@ def test_set_implicit_both_appengine_and_compute(self):
     def test_set_implicit_three_env_appengine_and_compute(self):
         from gcloud.datastore import _implicit_environ
 
-        IMPLICIT_DATASET_ID = 'IMPLICIT'
-        APP_IDENTITY = _AppIdentity('GAE')
-        connection = _HTTPConnection(200, 'GCE')
+        IMPLICIT_DATASET_ID = 's~IMPLICIT'
+        APP_IDENTITY = _AppIdentity('s~GAE')
+        connection = _HTTPConnection(200, 's~GCE')
 
         with self._monkeyEnviron(IMPLICIT_DATASET_ID):
             with self._monkeyImplicit(connection=connection,
@@ -210,6 +210,67 @@ def test_set_implicit_three_env_appengine_and_compute(self):
         self.assertEqual(connection.timeout, None)
 
 
+class Test__find_true_dataset_id(unittest2.TestCase):
+
+    def _callFUT(self, dataset_id, connection=None):
+        from gcloud.datastore import _find_true_dataset_id
+        return _find_true_dataset_id(dataset_id, connection=connection)
+
+    def test_prefixed(self):
+        PREFIXED = 's~DATASET'
+        result = self._callFUT(PREFIXED)
+        self.assertEqual(PREFIXED, result)
+
+    def test_unprefixed_no_connection(self):
+        from gcloud.datastore import _implicit_environ
+
+        UNPREFIXED = 'DATASET'
+        self.assertEqual(_implicit_environ.CONNECTION, None)
+        with self.assertRaises(AttributeError):
+            self._callFUT(UNPREFIXED)
+
+    def test_unprefixed_bogus_key_miss(self):
+        UNPREFIXED = 'DATASET'
+        PREFIX = 's~'
+        CONNECTION = _Connection(PREFIX, from_missing=False)
+        result = self._callFUT(UNPREFIXED, connection=CONNECTION)
+
+        self.assertEqual(CONNECTION._called_dataset_id, UNPREFIXED)
+
+        self.assertEqual(len(CONNECTION._lookup_result), 1)
+
+        # Make sure just one.
+        called_key_pb, = CONNECTION._called_key_pbs
+        path_element = called_key_pb.path_element
+        self.assertEqual(len(path_element), 1)
+        self.assertEqual(path_element[0].kind, '__MissingLookupKind')
+        self.assertEqual(path_element[0].id, 1)
+        self.assertFalse(path_element[0].HasField('name'))
+
+        PREFIXED = PREFIX + UNPREFIXED
+        self.assertEqual(result, PREFIXED)
+
+    def test_unprefixed_bogus_key_hit(self):
+        UNPREFIXED = 'DATASET'
+        PREFIX = 'e~'
+        CONNECTION = _Connection(PREFIX, from_missing=True)
+        result = self._callFUT(UNPREFIXED, connection=CONNECTION)
+
+        self.assertEqual(CONNECTION._called_dataset_id, UNPREFIXED)
+        self.assertEqual(CONNECTION._lookup_result, [])
+
+        # Make sure just one.
+        called_key_pb, = CONNECTION._called_key_pbs
+        path_element = called_key_pb.path_element
+        self.assertEqual(len(path_element), 1)
+        self.assertEqual(path_element[0].kind, '__MissingLookupKind')
+        self.assertEqual(path_element[0].id, 1)
+        self.assertFalse(path_element[0].HasField('name'))
+
+        PREFIXED = PREFIX + UNPREFIXED
+        self.assertEqual(result, PREFIXED)
+
+
 class Test_set_default_connection(unittest2.TestCase):
 
     def setUp(self):
@@ -350,3 +411,35 @@ class _TimeoutHTTPConnection(_BaseHTTPConnection):
     def getresponse(self):
         import socket
         raise socket.timeout('timed out')
+
+
+class _Connection(object):
+
+    _called_dataset_id = _called_key_pbs = _lookup_result = None
+
+    def __init__(self, prefix, from_missing=False):
+        self.prefix = prefix
+        self.from_missing = from_missing
+
+    def lookup(self, dataset_id, key_pbs):
+        from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
+
+        # Store the arguments called with.
+        self._called_dataset_id = dataset_id
+        self._called_key_pbs = key_pbs
+
+        key_pb, = key_pbs
+
+        response = datastore_pb.Entity()
+        response.key.CopyFrom(key_pb)
+        response.key.partition_id.dataset_id = self.prefix + dataset_id
+
+        missing = []
+        deferred = []
+        if self.from_missing:
+            missing[:] = [response]
+            self._lookup_result = []
+        else:
+            self._lookup_result = [response]
+
+        return self._lookup_result, missing, deferred
diff --git a/gcloud/datastore/test_api.py b/gcloud/datastore/test_api.py
index 587fb9448a0c..1ed92bee0ac2 100644
--- a/gcloud/datastore/test_api.py
+++ b/gcloud/datastore/test_api.py
@@ -306,10 +306,6 @@ def test_w_deferred_from_backend_but_not_passed(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
         from gcloud.datastore.connection import Connection
         from gcloud.datastore.key import Key
-        from gcloud.datastore import test_connection
-
-        # Shortening name, import line above was too long.
-        cmp_key_after_req = test_connection._compare_key_pb_after_request
 
         DATASET_ID = 'DATASET'
         key1 = Key('Kind', dataset_id=DATASET_ID)
@@ -372,15 +368,15 @@ def test_w_deferred_from_backend_but_not_passed(self):
         request.ParseFromString(cw[0]['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 2)
-        cmp_key_after_req(self, key_pb1, keys[0])
-        cmp_key_after_req(self, key_pb2, keys[1])
+        self.assertEqual(key_pb1, keys[0])
+        self.assertEqual(key_pb2, keys[1])
 
         # Make sure the second called with argument checks out.
         self._verifyProtobufCall(cw[1], URI, conn)
         request.ParseFromString(cw[1]['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 1)
-        cmp_key_after_req(self, key_pb2, keys[0])
+        self.assertEqual(key_pb2, keys[0])
 
     def test_hit(self):
         from gcloud.datastore.key import Key
diff --git a/gcloud/datastore/test_batch.py b/gcloud/datastore/test_batch.py
index 579fee92a612..709c8cd16b04 100644
--- a/gcloud/datastore/test_batch.py
+++ b/gcloud/datastore/test_batch.py
@@ -215,28 +215,9 @@ def test_put_entity_w_completed_key_prefixed_dataset_id(self):
         batch = self._makeOne(dataset_id=_DATASET, connection=connection)
         entity = _Entity(_PROPERTIES)
         entity.exclude_from_indexes = ('baz', 'spam')
-        key = entity.key = _Key('s~' + _DATASET)
+        entity.key = _Key('s~' + _DATASET)
 
-        batch.put(entity)
-
-        insert_auto_ids = list(batch.mutation.insert_auto_id)
-        self.assertEqual(len(insert_auto_ids), 0)
-        upserts = list(batch.mutation.upsert)
-        self.assertEqual(len(upserts), 1)
-
-        upsert = upserts[0]
-        self.assertEqual(upsert.key, key._key)
-        props = dict([(prop.name, prop.value) for prop in upsert.property])
-        self.assertTrue(props['foo'].indexed)
-        self.assertFalse(props['baz'].indexed)
-        self.assertTrue(props['spam'].indexed)
-        self.assertFalse(props['spam'].list_value[0].indexed)
-        self.assertFalse(props['spam'].list_value[1].indexed)
-        self.assertFalse(props['spam'].list_value[2].indexed)
-        self.assertFalse('frotz' in props)
-
-        deletes = list(batch.mutation.delete)
-        self.assertEqual(len(deletes), 0)
+        self.assertRaises(ValueError, batch.put, entity)
 
     def test_delete_w_partial_key(self):
         _DATASET = 'DATASET'
@@ -277,15 +258,7 @@ def test_delete_w_completed_key_w_prefixed_dataset_id(self):
         batch = self._makeOne(dataset_id=_DATASET, connection=connection)
         key = _Key('s~' + _DATASET)
 
-        batch.delete(key)
-
-        insert_auto_ids = list(batch.mutation.insert_auto_id)
-        self.assertEqual(len(insert_auto_ids), 0)
-        upserts = list(batch.mutation.upsert)
-        self.assertEqual(len(upserts), 0)
-        deletes = list(batch.mutation.delete)
-        self.assertEqual(len(deletes), 1)
-        self.assertEqual(deletes[0], key._key)
+        self.assertRaises(ValueError, batch.delete, key)
 
     def test_commit(self):
         _DATASET = 'DATASET'
diff --git a/gcloud/datastore/test_connection.py b/gcloud/datastore/test_connection.py
index 053e2e0c75b3..f52630b01b0a 100644
--- a/gcloud/datastore/test_connection.py
+++ b/gcloud/datastore/test_connection.py
@@ -205,7 +205,7 @@ def test_lookup_single_key_empty_response(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 1)
-        _compare_key_pb_after_request(self, key_pb, keys[0])
+        self.assertEqual(key_pb, keys[0])
 
     def test_lookup_single_key_empty_response_w_eventual(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
@@ -235,7 +235,7 @@ def test_lookup_single_key_empty_response_w_eventual(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 1)
-        _compare_key_pb_after_request(self, key_pb, keys[0])
+        self.assertEqual(key_pb, keys[0])
         self.assertEqual(request.read_options.read_consistency,
                          datastore_pb.ReadOptions.EVENTUAL)
         self.assertEqual(request.read_options.transaction, b'')
@@ -277,7 +277,7 @@ def test_lookup_single_key_empty_response_w_transaction(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 1)
-        _compare_key_pb_after_request(self, key_pb, keys[0])
+        self.assertEqual(key_pb, keys[0])
         self.assertEqual(request.read_options.transaction, TRANSACTION)
 
     def test_lookup_single_key_nonempty_response(self):
@@ -311,7 +311,7 @@ def test_lookup_single_key_nonempty_response(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 1)
-        _compare_key_pb_after_request(self, key_pb, keys[0])
+        self.assertEqual(key_pb, keys[0])
 
     def test_lookup_multiple_keys_empty_response(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
@@ -341,8 +341,8 @@ def test_lookup_multiple_keys_empty_response(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 2)
-        _compare_key_pb_after_request(self, key_pb1, keys[0])
-        _compare_key_pb_after_request(self, key_pb2, keys[1])
+        self.assertEqual(key_pb1, keys[0])
+        self.assertEqual(key_pb2, keys[1])
 
     def test_lookup_multiple_keys_w_missing(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
@@ -377,8 +377,8 @@ def test_lookup_multiple_keys_w_missing(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 2)
-        _compare_key_pb_after_request(self, key_pb1, keys[0])
-        _compare_key_pb_after_request(self, key_pb2, keys[1])
+        self.assertEqual(key_pb1, keys[0])
+        self.assertEqual(key_pb2, keys[1])
 
     def test_lookup_multiple_keys_w_deferred(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
@@ -415,8 +415,8 @@ def test_lookup_multiple_keys_w_deferred(self):
         request.ParseFromString(cw['body'])
         keys = list(request.key)
         self.assertEqual(len(keys), 2)
-        _compare_key_pb_after_request(self, key_pb1, keys[0])
-        _compare_key_pb_after_request(self, key_pb2, keys[1])
+        self.assertEqual(key_pb1, keys[0])
+        self.assertEqual(key_pb2, keys[1])
 
     def test_run_query_w_eventual_no_transaction(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
@@ -779,7 +779,7 @@ def test_allocate_ids_non_empty(self):
         request.ParseFromString(cw['body'])
         self.assertEqual(len(request.key), len(before_key_pbs))
         for key_before, key_after in zip(before_key_pbs, request.key):
-            _compare_key_pb_after_request(self, key_before, key_after)
+            self.assertEqual(key_before, key_after)
 
 
 class Http(object):
@@ -796,16 +796,6 @@ def request(self, **kw):
         return self._response, self._content
 
 
-def _compare_key_pb_after_request(test, key_before, key_after):
-    test.assertFalse(key_after.partition_id.HasField('dataset_id'))
-    test.assertEqual(key_before.partition_id.namespace,
-                     key_after.partition_id.namespace)
-    test.assertEqual(len(key_before.path_element),
-                     len(key_after.path_element))
-    for elt1, elt2 in zip(key_before.path_element, key_after.path_element):
-        test.assertEqual(elt1, elt2)
-
-
 class _Connection(object):
     _called_with = None
     _missing = _deferred = ()
diff --git a/gcloud/datastore/test_helpers.py b/gcloud/datastore/test_helpers.py
index bf2a5d676e69..5c6089d0afad 100644
--- a/gcloud/datastore/test_helpers.py
+++ b/gcloud/datastore/test_helpers.py
@@ -500,52 +500,3 @@ def test_list(self):
         self.assertEqual(marshalled[0].string_value, values[0])
         self.assertEqual(marshalled[1].integer_value, values[1])
         self.assertEqual(marshalled[2].double_value, values[2])
-
-
-class Test__prepare_key_for_request(unittest2.TestCase):
-
-    def _callFUT(self, key_pb):
-        from gcloud.datastore.helpers import _prepare_key_for_request
-
-        return _prepare_key_for_request(key_pb)
-
-    def test_prepare_dataset_id_valid(self):
-        from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
-        key = datastore_pb.Key()
-        key.partition_id.dataset_id = 'foo'
-        new_key = self._callFUT(key)
-        self.assertFalse(new_key is key)
-
-        key_without = datastore_pb.Key()
-        new_key.ClearField('partition_id')
-        self.assertEqual(new_key, key_without)
-
-    def test_prepare_dataset_id_unset(self):
-        from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
-        key = datastore_pb.Key()
-        new_key = self._callFUT(key)
-        self.assertTrue(new_key is key)
-
-
-class Test__dataset_ids_equal(unittest2.TestCase):
-
-    def _callFUT(self, dataset_id1, dataset_id2):
-        from gcloud.datastore.helpers import _dataset_ids_equal
-        return _dataset_ids_equal(dataset_id1, dataset_id2)
-
-    def test_identical_prefixed(self):
-        self.assertTrue(self._callFUT('s~foo', 's~foo'))
-        self.assertTrue(self._callFUT('e~bar', 'e~bar'))
-
-    def test_different_prefixed(self):
-        self.assertFalse(self._callFUT('s~foo', 's~bar'))
-        self.assertFalse(self._callFUT('s~foo', 'e~foo'))
-
-    def test_all_unprefixed(self):
-        self.assertTrue(self._callFUT('foo', 'foo'))
-        self.assertFalse(self._callFUT('foo', 'bar'))
-
-    def test_unprefixed_with_prefixed(self):
-        self.assertTrue(self._callFUT('foo', 's~foo'))
-        self.assertTrue(self._callFUT('foo', 'e~foo'))
-        self.assertFalse(self._callFUT('foo', 's~bar'))
diff --git a/gcloud/datastore/test_query.py b/gcloud/datastore/test_query.py
index c27deede9ff6..4f9921ff3dc5 100644
--- a/gcloud/datastore/test_query.py
+++ b/gcloud/datastore/test_query.py
@@ -573,7 +573,7 @@ def test_kind(self):
     def test_ancestor(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
         from gcloud.datastore.key import Key
-        from gcloud.datastore.helpers import _prepare_key_for_request
+
         ancestor = Key('Ancestor', 123, dataset_id='DATASET')
         pb = self._callFUT(_Query(ancestor=ancestor))
         cfilter = pb.filter.composite_filter
@@ -581,7 +581,7 @@ def test_ancestor(self):
         self.assertEqual(len(cfilter.filter), 1)
         pfilter = cfilter.filter[0].property_filter
         self.assertEqual(pfilter.property.name, '__key__')
-        ancestor_pb = _prepare_key_for_request(ancestor.to_protobuf())
+        ancestor_pb = ancestor.to_protobuf()
         self.assertEqual(pfilter.value.key_value, ancestor_pb)
 
     def test_filter(self):
@@ -601,7 +601,7 @@ def test_filter(self):
     def test_filter_key(self):
         from gcloud.datastore import _datastore_v1_pb2 as datastore_pb
         from gcloud.datastore.key import Key
-        from gcloud.datastore.helpers import _prepare_key_for_request
+
         key = Key('Kind', 123, dataset_id='DATASET')
         query = _Query(filters=[('__key__', '=', key)])
         query.OPERATORS = {
@@ -613,7 +613,7 @@ def test_filter_key(self):
         self.assertEqual(len(cfilter.filter), 1)
         pfilter = cfilter.filter[0].property_filter
         self.assertEqual(pfilter.property.name, '__key__')
-        key_pb = _prepare_key_for_request(key.to_protobuf())
+        key_pb = key.to_protobuf()
         self.assertEqual(pfilter.value.key_value, key_pb)
 
     def test_order(self):
diff --git a/regression/datastore.py b/regression/datastore.py
index 52a2739a43d1..c75aa9df3061 100644
--- a/regression/datastore.py
+++ b/regression/datastore.py
@@ -95,8 +95,8 @@ def _generic_test_post(self, name=None, key_id=None):
         # Check the keys are the same.
         self.assertEqual(retrieved_entity.key.path, entity.key.path)
         self.assertEqual(retrieved_entity.key.namespace, entity.key.namespace)
-        self.assertTrue(_compare_dataset_ids(
-            retrieved_entity.key.dataset_id, entity.key.dataset_id))
+        self.assertEqual(retrieved_entity.key.dataset_id,
+                         entity.key.dataset_id)
 
         # Check the data is the same.
         retrieved_dict = dict(retrieved_entity.items())
@@ -361,19 +361,3 @@ def test_transaction(self):
         retrieved_dict = dict(retrieved_entity.items())
         entity_dict = dict(entity.items())
         self.assertEqual(retrieved_dict, entity_dict)
-
-
-def _compare_dataset_ids(dataset_id1, dataset_id2):
-    if dataset_id1 == dataset_id2:
-        return True
-
-    if dataset_id1.startswith('s~') or dataset_id1.startswith('e~'):
-        # If `dataset_id1` is prefixed and not matching, then the only way
-        # they can match is if `dataset_id2` is unprefixed.
-        return dataset_id1[2:] == dataset_id2
-    elif dataset_id2.startswith('s~') or dataset_id2.startswith('e~'):
-        # Here we know `dataset_id1` is unprefixed and `dataset_id2`
-        # is prefixed.
-        return dataset_id1 == dataset_id2[2:]
-
-    return False