This commit is contained in:
Philipp Moritz committed 2017-08-26 20:40:31 -07:00
1 parent 20b119e3ce
commit 44c3b3dc53
4 files changed
+16 -19

No files matched your search

-9
View File
@@ -330,15 +330,6 @@ cdef class NativeFile:
cdef get_reader(object source, shared_ptr[RandomAccessFile]* reader)
cdef get_writer(object source, shared_ptr[OutputStream]* writer)
cdef class SerializedPyObject:
cdef:
CSerializedPyObject data
cdef readonly:
object base
cdef _write_to(self, OutputStream* stream)
cdef public object pyarrow_wrap_buffer(const shared_ptr[CBuffer]& buf)
cdef public object pyarrow_wrap_data_type(const shared_ptr[CDataType]& type)
cdef public object pyarrow_wrap_field(const shared_ptr[CField]& field)
+9 -8
View File
@@ -29,7 +29,7 @@ from cpython.pycapsule cimport *
import collections
import pyarrow
from pyarrow.lib cimport Buffer, NativeFile, check_status, SerializedPyObject
from pyarrow.lib cimport Buffer, NativeFile, check_status
from pyarrow.includes.libarrow cimport (CMutableBuffer, CBuffer,
CFixedSizeBufferWriter, CStatus)
@@ -616,7 +616,7 @@ def put(PlasmaClient client, value, object_id=None):
client : PlasmaClient
The client connected to the object store we put the value in.
value : object
A Python object to store.
A Python object to store. Currently only lists are supported.
object_id : ObjectID, default None
If this is provided, the specified object ID will be used to refer
to the object.
@@ -626,7 +626,7 @@ def put(PlasmaClient client, value, object_id=None):
The object ID associated to the Python object.
"""
cdef ObjectID id = object_id if object_id else ObjectID.from_random()
cdef SerializedPyObject serialized = pyarrow.serialize(value)
serialized = pyarrow.serialize(value)
buffer = client.create(id, serialized.total_bytes)
stream = pyarrow.FixedSizeBufferOutputStream(buffer)
stream.set_memcopy_threads(4)
@@ -642,8 +642,9 @@ def get(PlasmaClient client, object_ids, timeout_ms=-1):
----------
client : PlasmaClient
The client connected to the object store we get objects from.
object_ids : list
The list of object IDs associated to the values we get from the store.
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
@@ -651,9 +652,9 @@ def get(PlasmaClient client, object_ids, timeout_ms=-1):
Returns
-------
list
List of Python values for the data associated with the object_ids
and ObjectNotAvailable if the object was not available.
list or object
Python value or list of Python values for the data associated with
the object_ids and ObjectNotAvailable if the object was not available.
"""
if isinstance(object_ids, collections.Sequence):
results = []
+5
View File
@@ -135,6 +135,11 @@ cdef class SerializedPyObject:
"""
Arrow-serialized representation of Python object
"""
cdef:
CSerializedPyObject data
cdef readonly:
object base
property total_bytes:
+2 -2
View File
@@ -119,7 +119,7 @@ def start_plasma_store(plasma_store_memory=DEFAULT_PLASMA_STORE_MEMORY,
"""
if use_valgrind and use_profiler:
raise Exception("Cannot use valgrind and profiler at the same time.")
plasma_store_executable = "plasma_store"
plasma_store_executable = os.path.join(pa.__path__[0], "plasma_store")
plasma_store_name = "/tmp/plasma_store{}".format(random_name())
command = [plasma_store_executable,
"-s", plasma_store_name,
@@ -282,7 +282,7 @@ class TestPlasmaClient(object):
result = pa.plasma.get(self.plasma_client, object_id)
assert result == value
object_id = pa.plasma.ObjectID(np.bytes.random(20))
object_id = pa.plasma.ObjectID.from_random()
[result] = pa.plasma.get(self.plasma_client, [object_id], timeout_ms=0)
assert result == pa.plasma.ObjectNotAvailable