Update arrow to include user defined status for plasma (#5156)

bazel/ray_deps_setup.bzl

@@ -67,7 +67,7 @@ def ray_deps_setup():
     new_git_repository(
         name = "plasma",
         build_file = "@//bazel:BUILD.plasma",
-        commit = "aa9f08c2b927dee09f6193033f1678fb1d42114c",
+        commit = "d0d9ecec33413f7ef6c7f91448a802666ad5f871",
         remote = "https://github.com/apache/arrow",
     )
 

build.sh

@@ -102,7 +102,7 @@ pushd "$BUILD_DIR"
 # the commit listed in the command.
 $PYTHON_EXECUTABLE -m pip install \
     --target="$ROOT_DIR/python/ray/pyarrow_files" pyarrow==0.14.0.RAY \
-    --find-links https://s3-us-west-2.amazonaws.com/arrow-wheels/8772ea2ca8bbaf68576227991575ed0de3b23bff/index.html
+    --find-links https://s3-us-west-2.amazonaws.com/arrow-wheels/f86340a3b597502bacc801b17ab03c89d31aa561/index.html
 export PYTHON_BIN_PATH="$PYTHON_EXECUTABLE"
 
 if [ "$RAY_BUILD_JAVA" == "YES" ]; then

python/ray/worker.py

@@ -390,7 +390,7 @@ def put_object(self, object_id, value):
         # Serialize and put the object in the object store.
         try:
             self.store_and_register(object_id, value)
-        except pyarrow.PlasmaObjectExists:
+        except pyarrow.plasma.PlasmaObjectExists:
             # The object already exists in the object store, so there is no
             # need to add it again. TODO(rkn): We need to compare the hashes
             # and make sure that the objects are in fact the same. We also

src/ray/raylet/node_manager.cc

@@ -1421,7 +1421,7 @@ void NodeManager::TreatTaskAsFailed(const Task &task, const ErrorType &error_typ
   for (int64_t i = 0; i < num_returns; i++) {
     const auto object_id = spec.ReturnId(i).ToPlasmaId();
     arrow::Status status = store_client_.CreateAndSeal(object_id, "", meta);
-    if (!status.ok() && !status.IsPlasmaObjectExists()) {
+    if (!status.ok() && !plasma::IsPlasmaObjectExists(status)) {
       // If we failed to save the error code, log a warning and push an error message
       // to the driver.
       std::ostringstream stream;