You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ro...@apache.org on 2018/10/29 17:04:15 UTC
[arrow] branch master updated: ARROW-3555: [Plasma] Unify plasma
client get function using metadata.
This is an automated email from the ASF dual-hosted git repository.
robertnishihara pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new ce8bc4e ARROW-3555: [Plasma] Unify plasma client get function using metadata.
ce8bc4e is described below
commit ce8bc4e13fd5618688d51cdc9b9cc3a127a10837
Author: Yuhong Guo <yu...@antfin.com>
AuthorDate: Mon Oct 29 10:03:47 2018 -0700
ARROW-3555: [Plasma] Unify plasma client get function using metadata.
Sometimes, it is very hard for the data consumer to know whether an object is a buffer or other types of object. If we use `try-catch` statement to catch the pyarrow deserialization exception and then using `plasma_client.get_buffer`, the code is not clean.
As discussed with @robertnishihara , we may leverage the metadata which is not used at all to mark the buffer data. Furthermore, this will avoid output `None` when the object is actually not available, which is showed in the test.
In the client of other language, corresponding change would be easy.
Author: Yuhong Guo <yu...@antfin.com>
Author: Robert Nishihara <ro...@gmail.com>
Closes #2788 from guoyuhong/plasmaBytes and squashes the following commits:
4adbd17 <Robert Nishihara> Update _plasma.pyx
ea75fb4 <Yuhong Guo> Fix test error
fbb48ea <Yuhong Guo> Try to move get_raw_buffer to unified get_buffers
0d7ef1d <Yuhong Guo> lint
a7041df <Yuhong Guo> Fix test failures
47fb44e <Yuhong Guo> Add metadata support for plasma.
---
python/pyarrow/_plasma.pyx | 65 +++++++++++++------------------------
python/pyarrow/tests/test_plasma.py | 48 +++++++++++++++++++--------
2 files changed, 56 insertions(+), 57 deletions(-)
diff --git a/python/pyarrow/_plasma.pyx b/python/pyarrow/_plasma.pyx
index dfb6613..677e768 100644
--- a/python/pyarrow/_plasma.pyx
+++ b/python/pyarrow/_plasma.pyx
@@ -377,7 +377,7 @@ cdef class PlasmaClient:
check_status(self.client.get().CreateAndSeal(object_id.data, data,
metadata))
- def get_buffers(self, object_ids, timeout_ms=-1):
+ def get_buffers(self, object_ids, timeout_ms=-1, with_meta=False):
"""
Returns data buffer from the PlasmaStore based on object ID.
@@ -396,17 +396,28 @@ cdef class PlasmaClient:
Returns
-------
list
- List of PlasmaBuffers for the data associated with the object_ids
- and None if the object was not available.
+ If with_meta=False, this is a list of PlasmaBuffers for the data
+ associated with the object_ids and None if the object was not
+ available. If with_meta=True, this is a list of tuples of
+ PlasmaBuffer and metadata bytes.
"""
cdef c_vector[CObjectBuffer] object_buffers
self._get_object_buffers(object_ids, timeout_ms, &object_buffers)
result = []
for i in range(object_buffers.size()):
if object_buffers[i].data.get() != nullptr:
- result.append(pyarrow_wrap_buffer(object_buffers[i].data))
+ data = pyarrow_wrap_buffer(object_buffers[i].data)
else:
- result.append(None)
+ data = None
+ if not with_meta:
+ result.append(data)
+ else:
+ if object_buffers[i].metadata.get() != nullptr:
+ size = object_buffers[i].metadata.get().size()
+ metadata = object_buffers[i].metadata.get().data()[:size]
+ else:
+ metadata = None
+ result.append((metadata, data))
return result
def get_metadata(self, object_ids, timeout_ms=-1):
@@ -441,8 +452,8 @@ cdef class PlasmaClient:
result.append(None)
return result
- def put_buffer(self, object value, ObjectID object_id=None,
- int memcopy_threads=6):
+ def put_raw_buffer(self, object value, ObjectID object_id=None,
+ c_string metadata=b"", int memcopy_threads=6):
"""
Store Python buffer into the object store.
@@ -453,6 +464,9 @@ cdef class PlasmaClient:
object_id : ObjectID, default None
If this is provided, the specified object ID will be used to refer
to the object.
+ metadata : bytes
+ An optional string of bytes encoding whatever metadata the user
+ wishes to encode.
memcopy_threads : int, default 6
The number of threads to use to write the serialized object into
the object store for large objects.
@@ -464,48 +478,13 @@ cdef class PlasmaClient:
cdef ObjectID target_id = (object_id if object_id
else ObjectID.from_random())
cdef Buffer arrow_buffer = pyarrow.py_buffer(value)
- write_buffer = self.create(target_id, len(value))
+ write_buffer = self.create(target_id, len(value), metadata)
stream = pyarrow.FixedSizeBufferWriter(write_buffer)
stream.set_memcopy_threads(memcopy_threads)
stream.write(arrow_buffer)
self.seal(target_id)
return target_id
- def get_buffer(self, object_ids, int timeout_ms=-1):
- """
- Get one or more Python buffers from the object store.
-
- Parameters
- ----------
- object_ids : list or ObjectID
- Object ID or list of object IDs associated to the values we get
- from the store.
- timeout_ms : int, default -1
- The number of milliseconds that the get call should block before
- timing out and returning. Pass -1 if the call should block and 0
- if the call should return immediately.
-
- Returns
- -------
- list of buffer objects or a buffer object
- Python buffer or list of Python buffer for the data
- associated with the object_ids and ObjectNotAvailable if the
- object was not available.
- """
- cdef c_vector[CObjectBuffer] object_buffers
- if isinstance(object_ids, collections.Sequence):
- results = []
- self._get_object_buffers(object_ids, timeout_ms, &object_buffers)
- for i in range(object_buffers.size()):
- if object_buffers[i].data.get() != nullptr:
- size = object_buffers[i].data.get().size()
- results.append(object_buffers[i].data.get().data()[:size])
- else:
- results.append(None)
- return results
- else:
- return self.get_buffer([object_ids], timeout_ms)[0]
-
def put(self, object value, ObjectID object_id=None, int memcopy_threads=6,
serialization_context=None):
"""
diff --git a/python/pyarrow/tests/test_plasma.py b/python/pyarrow/tests/test_plasma.py
index 25a8aac..4b1aa10 100644
--- a/python/pyarrow/tests/test_plasma.py
+++ b/python/pyarrow/tests/test_plasma.py
@@ -220,7 +220,9 @@ class TestPlasmaClient(object):
self.plasma_client.create_and_seal(object_id, i * b'a', i * b'b')
for i in range(1000):
- assert self.plasma_client.get_buffer(object_ids[i]) == i * b'a'
+ [data_tuple] = self.plasma_client.get_buffers([object_ids[i]],
+ with_meta=True)
+ assert data_tuple[1].to_pybytes() == i * b'a'
assert (self.plasma_client.get_metadata(
[object_ids[i]])[0].to_pybytes()
== i * b'b')
@@ -284,11 +286,13 @@ class TestPlasmaClient(object):
# not sealed.
object_id = random_object_id()
self.plasma_client.create(object_id, 10, b"metadata")
- assert self.plasma_client.get_buffer(object_id, timeout_ms=0) is None
- assert self.plasma_client.get_buffer(object_id, timeout_ms=1) is None
+ assert self.plasma_client.get_buffers(
+ [object_id], timeout_ms=0, with_meta=True)[0][1] is None
+ assert self.plasma_client.get_buffers(
+ [object_id], timeout_ms=1, with_meta=True)[0][1] is None
self.plasma_client.seal(object_id)
- assert (self.plasma_client.get_buffer(object_id, timeout_ms=0) is
- not None)
+ assert self.plasma_client.get_buffers(
+ [object_id], timeout_ms=0, with_meta=True)[0][1]is not None
def test_buffer_lifetime(self):
# ARROW-2195
@@ -333,20 +337,36 @@ class TestPlasmaClient(object):
[result] = self.plasma_client.get([object_id], timeout_ms=0)
assert result == pa.plasma.ObjectNotAvailable
- def test_put_and_get_buffer_without_serialization(self):
+ def test_put_and_get_raw_buffer(self):
temp_id = random_object_id()
- for value in [b"This is a bytes obj", temp_id.binary(), 10 * b"\x00"]:
- object_id = self.plasma_client.put_buffer(value)
- [result] = self.plasma_client.get_buffer([object_id])
- assert result == value
+ use_meta = b"RAW"
+
+ def deserialize_or_output(data_tuple):
+ if data_tuple[0] == use_meta:
+ return data_tuple[1].to_pybytes()
+ else:
+ if data_tuple[1] is None:
+ return pa.plasma.ObjectNotAvailable
+ else:
+ return pa.deserialize(data_tuple[1])
- result = self.plasma_client.get_buffer(object_id)
+ for value in [b"Bytes Test", temp_id.binary(), 10 * b"\x00", 123]:
+ if isinstance(value, bytes):
+ object_id = self.plasma_client.put_raw_buffer(
+ value, metadata=use_meta)
+ else:
+ object_id = self.plasma_client.put(value)
+ [result] = self.plasma_client.get_buffers([object_id],
+ with_meta=True)
+ result = deserialize_or_output(result)
assert result == value
object_id = random_object_id()
- [result] = self.plasma_client.get_buffer(
- [object_id], timeout_ms=0)
- assert result is None
+ [result] = self.plasma_client.get_buffers([object_id],
+ timeout_ms=0,
+ with_meta=True)
+ result = deserialize_or_output(result)
+ assert result == pa.plasma.ObjectNotAvailable
def test_put_and_get_serialization_context(self):