Update Arrow Package with breaking changes (#2440)

* Merge the breaking change of Arrow Package.

* Fix typo

* Fix lint.

* put forward declarations into header

* fix

* add protocol.h

* fix linting

src/plasma/CMakeLists.txt

@@ -15,15 +15,46 @@ endif()
 include_directories("${ARROW_DIR}/cpp/src/")
 # include_directories("${CMAKE_CURRENT_LIST_DIR}/../")
 
+set(PLASMA_FBS_SRC "${CMAKE_CURRENT_LIST_DIR}/format/plasma.fbs" "${CMAKE_CURRENT_LIST_DIR}/format/common.fbs")
+set(OUTPUT_DIR ${CMAKE_CURRENT_LIST_DIR}/format/)
+
+set(PLASMA_FBS_OUTPUT_FILES
+  "${OUTPUT_DIR}/plasma_generated.h"
+  "${OUTPUT_DIR}/common_generated.h")
+
+add_custom_target(gen_plasma_fbs DEPENDS ${PLASMA_FBS_OUTPUT_FILES})
+add_dependencies(gen_plasma_fbs flatbuffers_ep)
+
+# Copy the fbs files from Arrow project to local directory.
+add_custom_command(
+  OUTPUT ${PLASMA_FBS_SRC}
+  COMMAND cp -rf ${CMAKE_CURRENT_LIST_DIR}/../../thirdparty/build/arrow/cpp/src/plasma/format/ ${CMAKE_CURRENT_LIST_DIR}/format/
+  COMMENT "Copying ${PLASMA_FBS_SRC} to local"
+  VERBATIM)
+
+# Compile flatbuffers
+add_custom_command(
+  OUTPUT ${PLASMA_FBS_OUTPUT_FILES}
+  # The --gen-object-api flag generates a C++ class MessageT for each
+  # flatbuffers message Message, which can be used to store deserialized
+  # messages in data structures. This is currently used for ObjectInfo for
+  # example.
+  COMMAND ${FLATBUFFERS_COMPILER} -c -o ${OUTPUT_DIR} ${PLASMA_FBS_SRC} --gen-object-api --scoped-enums
+  DEPENDS ${PLASMA_FBS_SRC}
+  COMMENT "Running flatc compiler on ${PLASMA_FBS_SRC}"
+  VERBATIM)
+
 include_directories("${FLATBUFFERS_INCLUDE_DIR}")
 
 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC")
 
 add_executable(plasma_manager
   plasma_manager.cc)
+add_dependencies(plasma_manager gen_plasma_fbs)
 
 target_link_libraries(plasma_manager common ${PLASMA_STATIC_LIB} ray_static ${ARROW_STATIC_LIB} -lpthread ${Boost_SYSTEM_LIBRARY})
 
 define_test(client_tests "")
 define_test(manager_tests "" plasma_manager.cc)
 target_link_libraries(manager_tests ${Boost_SYSTEM_LIBRARY})
+add_dependencies(manager_tests gen_plasma_fbs)

src/plasma/format/plasma.fbs

@@ -1,274 +0,0 @@
-// Plasma protocol specification
-
-enum MessageType:int {
-  // Create a new object.
-  PlasmaCreateRequest = 1,
-  PlasmaCreateReply,
-  // Seal an object.
-  PlasmaSealRequest,
-  PlasmaSealReply,
-  // Get an object that is stored on the local Plasma store.
-  PlasmaGetRequest,
-  PlasmaGetReply,
-  // Release an object.
-  PlasmaReleaseRequest,
-  PlasmaReleaseReply,
-  // Delete an object.
-  PlasmaDeleteRequest,
-  PlasmaDeleteReply,
-  // Get status of an object.
-  PlasmaStatusRequest,
-  PlasmaStatusReply,
-  // See if the store contains an object (will be deprecated).
-  PlasmaContainsRequest,
-  PlasmaContainsReply,
-  // Get information for a newly connecting client.
-  PlasmaConnectRequest,
-  PlasmaConnectReply,
-  // Make room for new objects in the plasma store.
-  PlasmaEvictRequest,
-  PlasmaEvictReply,
-  // Fetch objects from remote Plasma stores.
-  PlasmaFetchRequest,
-  // Wait for objects to be ready either from local or remote Plasma stores.
-  PlasmaWaitRequest,
-  PlasmaWaitReply,
-  // Subscribe to a list of objects or to all objects.
-  PlasmaSubscribeRequest,
-  // Unsubscribe.
-  PlasmaUnsubscribeRequest,
-  // Sending and receiving data.
-  // PlasmaDataRequest initiates sending the data, there will be one
-  // such message per data transfer.
-  PlasmaDataRequest,
-  // PlasmaDataReply contains the actual data and is sent back to the
-  // object store that requested the data. For each transfer, multiple
-  // reply messages get sent. Each one contains a fixed number of bytes.
-  PlasmaDataReply,
-  // Object notifications.
-  PlasmaNotification
-}
-
-enum PlasmaError:int {
-  // Operation was successful.
-  OK,
-  // Trying to create an object that already exists.
-  ObjectExists,
-  // Trying to access an object that doesn't exist.
-  ObjectNonexistent,
-  // Trying to create an object but there isn't enough space in the store.
-  OutOfMemory
-}
-
-// Plasma store messages
-
-struct PlasmaObjectSpec {
-  // Index of the memory segment (= memory mapped file) that
-  // this object is allocated in.
-  segment_index: int;
-  // Size in bytes of this segment (needed to call mmap).
-  mmap_size: ulong;
-  // The offset in bytes in the memory mapped file of the data.
-  data_offset: ulong;
-  // The size in bytes of the data.
-  data_size: ulong;
-  // The offset in bytes in the memory mapped file of the metadata.
-  metadata_offset: ulong;
-  // The size in bytes of the metadata.
-  metadata_size: ulong;
-}
-
-table PlasmaCreateRequest {
-  // ID of the object to be created.
-  object_id: string;
-  // The size of the object's data in bytes.
-  data_size: ulong;
-  // The size of the object's metadata in bytes.
-  metadata_size: ulong;
-}
-
-table PlasmaCreateReply {
-  // ID of the object that was created.
-  object_id: string;
-  // The object that is returned with this reply.
-  plasma_object: PlasmaObjectSpec;
-  // Error that occurred for this call.
-  error: PlasmaError;
-}
-
-table PlasmaSealRequest {
-  // ID of the object to be sealed.
-  object_id: string;
-  // Hash of the object data.
-  digest: string;
-}
-
-table PlasmaSealReply {
-  // ID of the object that was sealed.
-  object_id: string;
-  // Error code.
-  error: PlasmaError;
-}
-
-table PlasmaGetRequest {
-  // IDs of the objects stored at local Plasma store we are getting.
-  object_ids: [string];
-  // The number of milliseconds before the request should timeout.
-  timeout_ms: long;
-}
-
-table PlasmaGetReply {
-  // IDs of the objects being returned.
-  // This number can be smaller than the number of requested
-  // objects if not all requested objects are stored and sealed
-  // in the local Plasma store.
-  object_ids: [string];
-  // Plasma object information, in the same order as their IDs.
-  plasma_objects: [PlasmaObjectSpec];
-  // The number of elements in both object_ids and plasma_objects arrays must agree.
-}
-
-table PlasmaReleaseRequest {
-  // ID of the object to be released.
-  object_id: string;
-}
-
-table PlasmaReleaseReply {
-  // ID of the object that was released.
-  object_id: string;
-  // Error code.
-  error: PlasmaError;
-}
-
-table PlasmaDeleteRequest {
-  // ID of the object to be deleted.
-  object_id: string;
-}
-
-table PlasmaDeleteReply {
-  // ID of the object that was deleted.
-  object_id: string;
-  // Error code.
-  error: PlasmaError;
-}
-
-table PlasmaStatusRequest {
-  // IDs of the objects stored at local Plasma store we request the status of.
-  object_ids: [string];
-}
-
-enum ObjectStatus:int {
-  // Object is stored in the local Plasma Store.
-  Local = 1,
-  // Object is stored on a remote Plasma store, and it is not stored on the
-  // local Plasma Store.
-  Remote,
-  // Object is not stored in the system.
-  Nonexistent,
-  // Object is currently transferred from a remote Plasma store the the local
-  // Plasma Store.
-  Transfer
-}
-
-table PlasmaStatusReply {
-  // IDs of the objects being returned.
-  object_ids: [string];
-  // Status of the object.
-  status: [ObjectStatus];
-}
-
-// PlasmaContains is a subset of PlasmaStatus which does not
-// involve the plasma manager, only the store. We should consider
-// unifying them in the future and deprecating PlasmaContains.
-
-table PlasmaContainsRequest {
-  // ID of the object we are querying.
-  object_id: string;
-}
-
-table PlasmaContainsReply {
-  // ID of the object we are querying.
-  object_id: string;
-  // 1 if the object is in the store and 0 otherwise.
-  has_object: int;
-}
-
-// PlasmaConnect is used by a plasma client the first time it connects with the
-// store. This is not really necessary, but is used to get some information
-// about the store such as its memory capacity.
-
-table PlasmaConnectRequest {
-}
-
-table PlasmaConnectReply {
-  // The memory capacity of the store.
-  memory_capacity: long;
-}
-
-table PlasmaEvictRequest {
-  // Number of bytes that shall be freed.
-  num_bytes: ulong;
-}
-
-table PlasmaEvictReply {
-  // Number of bytes that have been freed.
-  num_bytes: ulong;
-}
-
-table PlasmaFetchRequest {
-  // IDs of objects to be gotten.
-  object_ids: [string];
-}
-
-table ObjectRequestSpec {
-  // ID of the object.
-  object_id: string;
-  // The type of the object. This specifies whether we
-  // will be waiting for an object store in the local or
-  // global Plasma store.
-  type: int;
-}
-
-table PlasmaWaitRequest {
-  // Array of object requests whose status we are asking for.
-  object_requests: [ObjectRequestSpec];
-  // Number of objects expected to be returned, if available.
-  num_ready_objects: int;
-  // timeout
-  timeout: long;
-}
-
-table ObjectReply {
-  // ID of the object.
-  object_id: string;
-  // The object status. This specifies where the object is stored.
-  status: int;
-}
-
-table PlasmaWaitReply {
-  // Array of object requests being returned.
-  object_requests: [ObjectReply];
-  // Number of objects expected to be returned, if available.
-  num_ready_objects: int;
-}
-
-table PlasmaSubscribeRequest {
-}
-
-table PlasmaDataRequest {
-  // ID of the object that is requested.
-  object_id: string;
-  // The host address where the data shall be sent to.
-  address: string;
-  // The port of the manager the data shall be sent to.
-  port: int;
-}
-
-table PlasmaDataReply {
-  // ID of the object that will be sent.
-  object_id: string;
-  // Size of the object data in bytes.
-  object_size: ulong;
-  // Size of the metadata in bytes.
-  metadata_size: ulong;
-}

src/plasma/plasma_manager.cc

@@ -26,22 +26,24 @@
 #include <unordered_set>
 #include <vector>
 
+#include "common.h"
 #include "common_protocol.h"
+#include "event_loop.h"
+#include "format/plasma_generated.h"
 #include "io.h"
 #include "net.h"
-#include "event_loop.h"
-#include "common.h"
-#include "plasma/plasma.h"
-#include "plasma/events.h"
-#include "plasma/protocol.h"
 #include "plasma/client.h"
+#include "plasma/events.h"
 #include "plasma_manager.h"
-#include "state/db.h"
-#include "state/object_table.h"
-#include "state/error_table.h"
-#include "state/task_table.h"
-#include "state/db_client_table.h"
 #include "ray/gcs/client.h"
+#include "state/db.h"
+#include "state/db_client_table.h"
+#include "state/error_table.h"
+#include "state/object_table.h"
+#include "state/task_table.h"
+
+using plasma::ObjectLocation;
+using plasma::flatbuf::MessageType;
 
 int handle_sigpipe(plasma::Status s, int fd) {
   if (s.ok()) {
@@ -101,9 +103,9 @@ void process_status_request(ClientConnection *client_conn, ObjectID object_id);
  * @param context Client connection.
  * @return Status of object_id as defined in plasma.h
  */
-ObjectStatus request_status(ObjectID object_id,
-                            const std::vector<DBClientID> &manager_vector,
-                            void *context);
+ObjectLocation request_status(ObjectID object_id,
+                              const std::vector<DBClientID> &manager_vector,
+                              void *context);
 
 /**
  * Send requested object_id back to the Plasma Manager identified
@@ -394,7 +396,7 @@ void return_from_wait(PlasmaManagerState *manager_state,
 void update_object_wait_requests(PlasmaManagerState *manager_state,
                                  ObjectID obj_id,
                                  plasma::ObjectRequestType type,
-                                 ObjectStatus status) {
+                                 ObjectLocation status) {
   auto &object_wait_requests =
       object_wait_requests_from_type(manager_state, type);
   /* Update the in-progress wait requests in the specified table. */
@@ -418,9 +420,11 @@ void update_object_wait_requests(PlasmaManagerState *manager_state,
       /* Check that we found the object. */
       RAY_CHECK(object_request != wait_req->object_requests.end());
       /* Check that the object found was not previously known to us. */
-      RAY_CHECK(object_request->second.status == ObjectStatus::Nonexistent);
+      RAY_CHECK(object_request->second.location ==
+                plasma::ObjectLocation::Nonexistent);
       /* Update the found object's status to a known status. */
-      object_request->second.status = status;
+      object_request->second.location =
+          static_cast<plasma::ObjectLocation>(status);
 
       /* If this wait request is done, reply to the client. */
       if (wait_req->num_satisfied == wait_req->num_objects_to_wait_for) {
@@ -1087,7 +1091,7 @@ void object_table_subscribe_callback(ObjectID object_id,
   /* Run the callback for wait requests. */
   update_object_wait_requests(manager_state, object_id,
                               plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE,
-                              ObjectStatus::Remote);
+                              ObjectLocation::Remote);
 }
 
 void process_fetch_requests(ClientConnection *client_conn,
@@ -1168,7 +1172,7 @@ void process_wait_request(ClientConnection *client_conn,
     /* Check if this object is already present locally. If so, mark the object
      * as present. */
     if (is_object_local(manager_state, obj_id)) {
-      object_request.status = ObjectStatus::Local;
+      object_request.location = plasma::ObjectLocation::Local;
       wait_req->num_satisfied += 1;
       continue;
     }
@@ -1238,28 +1242,28 @@ void request_status_done(ObjectID object_id,
       client_conn->fd);
 }
 
-ObjectStatus request_status(ObjectID object_id,
-                            const std::vector<DBClientID> &manager_vector,
-                            void *context) {
+ObjectLocation request_status(ObjectID object_id,
+                              const std::vector<DBClientID> &manager_vector,
+                              void *context) {
   ClientConnection *client_conn = (ClientConnection *) context;
 
   /* Return success immediately if we already have this object. */
   if (is_object_local(client_conn->manager_state, object_id)) {
-    return ObjectStatus::Local;
+    return ObjectLocation::Local;
   }
 
   /* Since object is not stored at the local locally, manager_vector.size() > 0
    * means that the object is stored at another remote object. Otherwise, if
    * manager_vector.size() == 0, the object is not stored anywhere. */
-  return manager_vector.size() > 0 ? ObjectStatus::Remote
-                                   : ObjectStatus::Nonexistent;
+  return manager_vector.size() > 0 ? ObjectLocation::Remote
+                                   : ObjectLocation::Nonexistent;
 }
 
 void object_table_lookup_fail_callback(ObjectID object_id,
                                        void *user_context,
                                        void *user_data) {
-  /* Fail for now. Later, we may want to send a ObjectStatus::Nonexistent to the
-   * client. */
+  /* Fail for now. Later, we may want to send a ObjectLocation::Nonexistent to
+   * the client. */
   RAY_CHECK(0);
 }
 
@@ -1267,7 +1271,7 @@ void process_status_request(ClientConnection *client_conn,
                             plasma::ObjectID object_id) {
   /* Return success immediately if we already have this object. */
   if (is_object_local(client_conn->manager_state, object_id)) {
-    int status = static_cast<int>(ObjectStatus::Local);
+    int status = static_cast<int>(ObjectLocation::Local);
     handle_sigpipe(
         plasma::SendStatusReply(client_conn->fd, &object_id, &status, 1),
         client_conn->fd);
@@ -1275,7 +1279,7 @@ void process_status_request(ClientConnection *client_conn,
   }
 
   if (client_conn->manager_state->db == NULL) {
-    auto status = static_cast<int>(ObjectStatus::Nonexistent);
+    auto status = static_cast<int>(ObjectLocation::Nonexistent);
     handle_sigpipe(
         plasma::SendStatusReply(client_conn->fd, &object_id, &status, 1),
         client_conn->fd);
@@ -1374,10 +1378,10 @@ void process_add_object_notification(PlasmaManagerState *state,
   /* Update the in-progress local and remote wait requests. */
   update_object_wait_requests(state, object_id,
                               plasma::ObjectRequestType::PLASMA_QUERY_LOCAL,
-                              ObjectStatus::Local);
+                              ObjectLocation::Local);
   update_object_wait_requests(state, object_id,
                               plasma::ObjectRequestType::PLASMA_QUERY_ANYWHERE,
-                              ObjectStatus::Local);
+                              ObjectLocation::Local);
 }
 
 void process_object_notification(event_loop *loop,

src/plasma/plasma_manager.h

@@ -1,6 +1,8 @@
 #ifndef PLASMA_MANAGER_H
 #define PLASMA_MANAGER_H
 
+#include "protocol.h"
+
 #ifndef RAY_NUM_RETRIES
 #define NUM_RETRIES 5
 #else
@@ -9,7 +11,6 @@
 
 typedef struct PlasmaManagerState PlasmaManagerState;
 typedef struct ClientConnection ClientConnection;
-enum class MessageType : int64_t;
 
 /**
  * Initializes the plasma manager state. This connects the manager to the local
@@ -155,7 +156,7 @@ ClientConnection *ClientConnection_listen(event_loop *loop,
 
 /* Buffer for requests between plasma managers. */
 typedef struct PlasmaRequestBuffer {
-  MessageType type;
+  plasma::flatbuf::MessageType type;
   ray::ObjectID object_id;
   uint8_t *data;
   int64_t data_size;

src/plasma/protocol.h

@@ -0,0 +1,77 @@
+#ifndef PLASMA_PROTOCOL_H
+#define PLASMA_PROTOCOL_H
+
+#include "./format/common_generated.h"
+#include "./format/plasma_generated.h"
+
+namespace plasma {
+
+namespace flatbuf {
+enum class MessageType : int64_t;
+};
+
+using arrow::Status;
+
+typedef std::unordered_map<ObjectID, ObjectRequest> ObjectRequestMap;
+
+Status PlasmaReceive(int sock,
+                     flatbuf::MessageType message_type,
+                     std::vector<uint8_t> *buffer);
+
+Status SendWaitReply(int sock,
+                     const ObjectRequestMap &object_requests,
+                     int num_ready_objects);
+
+Status SendStatusReply(int sock,
+                       ObjectID object_ids[],
+                       int object_status[],
+                       int64_t num_objects);
+
+Status SendDataRequest(int sock,
+                       ObjectID object_id,
+                       const char *address,
+                       int port);
+
+Status SendDataReply(int sock,
+                     ObjectID object_id,
+                     int64_t object_size,
+                     int64_t metadata_size);
+
+Status ReadDataRequest(uint8_t *data,
+                       size_t size,
+                       ObjectID *object_id,
+                       char **address,
+                       int *port);
+
+Status ReadDataReply(uint8_t *data,
+                     size_t size,
+                     ObjectID *object_id,
+                     int64_t *object_size,
+                     int64_t *metadata_size);
+
+Status ReadFetchRequest(uint8_t *data,
+                        size_t size,
+                        std::vector<ObjectID> &object_ids);
+
+Status ReadStatusRequest(uint8_t *data,
+                         size_t size,
+                         ObjectID object_ids[],
+                         int64_t num_objects);
+
+Status ReadWaitRequest(uint8_t *data,
+                       size_t size,
+                       ObjectRequestMap &object_requests,
+                       int64_t *timeout_ms,
+                       int *num_ready_objects);
+
+Status ReadStatusRequest(uint8_t *data,
+                         size_t size,
+                         ObjectID object_ids[],
+                         int64_t num_objects);
+
+std::unique_ptr<uint8_t[]> CreateObjectInfoBuffer(
+    flatbuf::ObjectInfoT *object_info);
+
+}  // namespace plasma
+
+#endif

src/plasma/test/client_tests.cc

@@ -5,8 +5,6 @@
 #include <sys/time.h>
 
 #include "plasma/common.h"
-#include "plasma/plasma.h"
-#include "plasma/protocol.h"
 #include "plasma/client.h"
 
 using namespace plasma;
@@ -25,7 +23,7 @@ TEST plasma_status_tests(void) {
   /* Test for object non-existence. */
   int status;
   ARROW_CHECK_OK(client1.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Nonexistent));
+  ASSERT(status == static_cast<int>(ObjectLocation::Nonexistent));
 
   /* Test for the object being in local Plasma store. */
   /* First create object. */
@@ -40,11 +38,11 @@ TEST plasma_status_tests(void) {
    */
   sleep(1);
   ARROW_CHECK_OK(client1.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Local));
+  ASSERT(status == static_cast<int>(ObjectLocation::Local));
 
   /* Test for object being remote. */
   ARROW_CHECK_OK(client2.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Remote));
+  ASSERT(status == static_cast<int>(ObjectLocation::Remote));
 
   ARROW_CHECK_OK(client1.Disconnect());
   ARROW_CHECK_OK(client2.Disconnect());
@@ -66,7 +64,7 @@ TEST plasma_fetch_tests(void) {
 
   /* No object in the system */
   ARROW_CHECK_OK(client1.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Nonexistent));
+  ASSERT(status == static_cast<int>(ObjectLocation::Nonexistent));
 
   /* Test for the object being in local Plasma store. */
   /* First create object. */
@@ -84,24 +82,24 @@ TEST plasma_fetch_tests(void) {
   ObjectID oid_array1[1] = {oid1};
   ARROW_CHECK_OK(client1.Fetch(1, oid_array1));
   ARROW_CHECK_OK(client1.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Local) ||
-         status == static_cast<int>(ObjectStatus::Nonexistent));
+  ASSERT(status == static_cast<int>(ObjectLocation::Local) ||
+         status == static_cast<int>(ObjectLocation::Nonexistent));
 
   /* Sleep to make sure Plasma Manager got the notification. */
   sleep(1);
   ARROW_CHECK_OK(client1.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Local));
+  ASSERT(status == static_cast<int>(ObjectLocation::Local));
 
   /* Test for object being remote. */
   ARROW_CHECK_OK(client2.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Remote));
+  ASSERT(status == static_cast<int>(ObjectLocation::Remote));
 
   /* Sleep to make sure the object has been fetched and it is now stored in the
    * local Plasma Store. */
   ARROW_CHECK_OK(client2.Fetch(1, oid_array1));
   sleep(1);
   ARROW_CHECK_OK(client2.Info(oid1, &status));
-  ASSERT(status == static_cast<int>(ObjectStatus::Local));
+  ASSERT(status == static_cast<int>(ObjectLocation::Local));
 
   sleep(1);
   ARROW_CHECK_OK(client1.Disconnect());

src/plasma/test/manager_tests.cc

@@ -14,10 +14,11 @@
 #include "event_loop.h"
 #include "io.h"
 
-#include "plasma/plasma.h"
-#include "plasma/client.h"
 #include "../plasma_manager.h"
-#include "plasma/protocol.h"
+#include "plasma/client.h"
+#include "../protocol.h"
+
+namespace fb = plasma::flatbuf;
 
 SUITE(plasma_manager_tests);
 
@@ -112,7 +113,7 @@ void destroy_plasma_mock(plasma_mock *mock) {
  * - Buffer a transfer request for the remote manager.
  * - Start and stop the event loop to make sure that we send the buffered
  *   request.
- * - Expect to see a MessageType::PlasmaDataRequest message on the remote
+ * - Expect to see a fb::MessageType::PlasmaDataRequest message on the remote
  *   manager with the correct object ID.
  */
 TEST request_transfer_test(void) {
@@ -128,8 +129,8 @@ TEST request_transfer_test(void) {
   event_loop_run(local_mock->loop);
   int read_fd = get_client_sock(remote_mock->read_conn);
   std::vector<uint8_t> request_data;
-  ARROW_CHECK_OK(plasma::PlasmaReceive(read_fd, MessageType::PlasmaDataRequest,
-                                       &request_data));
+  ARROW_CHECK_OK(plasma::PlasmaReceive(
+      read_fd, fb::MessageType::PlasmaDataRequest, &request_data));
   plasma::ObjectID object_id2;
   char *address;
   int port;
@@ -152,8 +153,8 @@ TEST request_transfer_test(void) {
  * - Buffer a transfer request for the remote managers.
  * - Start and stop the event loop after a timeout to make sure that we
  *   trigger the timeout on the first manager.
- * - Expect to see a MessageType::PlasmaDataRequest message on the second remote
- *   manager with the correct object ID.
+ * - Expect to see a fb::MessageType::PlasmaDataRequest message on the second
+ *   remote manager with the correct object ID.
  */
 TEST request_transfer_retry_test(void) {
   plasma_mock *local_mock = init_plasma_mock(NULL);
@@ -180,8 +181,8 @@ TEST request_transfer_retry_test(void) {
 
   int read_fd = get_client_sock(remote_mock2->read_conn);
   std::vector<uint8_t> request_data;
-  ARROW_CHECK_OK(plasma::PlasmaReceive(read_fd, MessageType::PlasmaDataRequest,
-                                       &request_data));
+  ARROW_CHECK_OK(plasma::PlasmaReceive(
+      read_fd, fb::MessageType::PlasmaDataRequest, &request_data));
   plasma::ObjectID object_id2;
   char *address;
   int port;
@@ -211,7 +212,7 @@ TEST read_write_object_chunk_test(void) {
   const int data_size = strlen(data) + 1;
   const int metadata_size = 0;
   PlasmaRequestBuffer remote_buf;
-  remote_buf.type = MessageType::PlasmaDataReply;
+  remote_buf.type = fb::MessageType::PlasmaDataReply;
   remote_buf.object_id = object_id;
   remote_buf.data = (uint8_t *) data;
   remote_buf.data_size = data_size;
@@ -255,7 +256,7 @@ TEST object_notifications_test(void) {
   RAY_CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0);
 
   ObjectID object_id = ObjectID::from_random();
-  ObjectInfoT info;
+  fb::ObjectInfoT info;
   info.object_id = object_id.binary();
   info.data_size = 10;
   info.metadata_size = 1;
@@ -269,7 +270,7 @@ TEST object_notifications_test(void) {
   ASSERT(!is_local);
 
   /* Check that the object is local after receiving an object notification. */
-  auto notification = plasma::create_object_info_buffer(&info);
+  auto notification = plasma::CreateObjectInfoBuffer(&info);
   int64_t size = *((int64_t *) notification.get());
   send(fd[1], notification.get(), sizeof(int64_t) + size, 0);
   process_object_notification(local_mock->loop, fd[0], local_mock->state, 0);
@@ -279,7 +280,7 @@ TEST object_notifications_test(void) {
   /* Check that the object is not local after receiving a notification about
    * the object deletion. */
   info.is_deletion = true;
-  notification = plasma::create_object_info_buffer(&info);
+  notification = plasma::CreateObjectInfoBuffer(&info);
   size = *((int64_t *) notification.get());
   send(fd[1], notification.get(), sizeof(int64_t) + size, 0);
   process_object_notification(local_mock->loop, fd[0], local_mock->state, 0);

src/ray/object_manager/object_buffer_pool.h

@@ -12,7 +12,6 @@
 
 #include "plasma/client.h"
 #include "plasma/events.h"
-#include "plasma/plasma.h"
 
 #include "ray/id.h"
 #include "ray/status.h"

src/ray/object_manager/object_manager.h

@@ -14,7 +14,6 @@
 
 #include "plasma/client.h"
 #include "plasma/events.h"
-#include "plasma/plasma.h"
 
 #include "ray/common/client_connection.h"
 #include "ray/id.h"

src/ray/object_manager/object_store_notification_manager.h

@@ -11,7 +11,6 @@
 
 #include "plasma/client.h"
 #include "plasma/events.h"
-#include "plasma/plasma.h"
 
 #include "ray/id.h"
 #include "ray/status.h"

thirdparty/scripts/build_arrow.sh

@@ -40,10 +40,10 @@ else
   exit 1
 fi
 
-# The PR for this commit is https://github.com/apache/arrow/pull/2104. We
+# The PR for this commit is https://github.com/apache/arrow/pull/2282. We
 # include the link here to make it easier to find the right commit because
 # Arrow often rewrites git history and invalidates certain commits.
-TARGET_COMMIT_ID=cecbcf7de00d2bf255bdba97cee1d37130e5bb79
+TARGET_COMMIT_ID=35ef303ad4f5a1f7a7e156e94ef331b7f9586ca5
 build_arrow() {
   echo "building arrow"