Integrate plasma store list facility (#2752)

src/common/format/common.fbs

@@ -65,6 +65,9 @@ table TaskInfo {
 }
 
 // Object information data structure.
+// NOTE(pcm): This structure is replicated in
+// https://github.com/apache/arrow/blob/master/cpp/src/plasma/format/common.fbs,
+// so if you modify it, you should also modify that one.
 table ObjectInfo {
   // Object ID of this object.
   object_id: string;
@@ -72,11 +75,14 @@ table ObjectInfo {
   data_size: long;
   // Number of bytes the metadata of this object occupies in memory.
   metadata_size: long;
+  // Number of clients using the objects.
+  ref_count: int;
   // Unix epoch of when this object was created.
   create_time: long;
   // How long creation of this object took.
   construct_duration: long;
-  // Hash of the object content.
+  // Hash of the object content. If the object is not sealed yet this is
+  // an empty string.
   digest: string;
   // Specifies if this object was deleted or added.
   is_deletion: bool;

src/plasma/test/client_tests.cc

@@ -4,6 +4,8 @@
 #include <unistd.h>
 #include <sys/time.h>
 
+#include "plasma/test-util.h"
+
 #include "plasma/common.h"
 #include "plasma/client.h"
 
@@ -18,7 +20,7 @@ TEST plasma_status_tests(void) {
   PlasmaClient client2;
   ARROW_CHECK_OK(client2.Connect("/tmp/store2", "/tmp/manager2",
                                  plasma::kPlasmaDefaultReleaseDelay));
-  ObjectID oid1 = ObjectID::from_random();
+  ObjectID oid1 = random_object_id();
 
   /* Test for object non-existence. */
   int status;
@@ -57,7 +59,7 @@ TEST plasma_fetch_tests(void) {
   PlasmaClient client2;
   ARROW_CHECK_OK(client2.Connect("/tmp/store2", "/tmp/manager2",
                                  plasma::kPlasmaDefaultReleaseDelay));
-  ObjectID oid1 = ObjectID::from_random();
+  ObjectID oid1 = random_object_id();
 
   /* Test for object non-existence. */
   int status;
@@ -129,7 +131,7 @@ TEST plasma_nonblocking_get_tests(void) {
   PlasmaClient client;
   ARROW_CHECK_OK(client.Connect("/tmp/store1", "/tmp/manager1",
                                 plasma::kPlasmaDefaultReleaseDelay));
-  ObjectID oid = ObjectID::from_random();
+  ObjectID oid = random_object_id();
   ObjectID oid_array[1] = {oid};
   ObjectBuffer obj_buffer;
 
@@ -165,8 +167,8 @@ TEST plasma_wait_for_objects_tests(void) {
   PlasmaClient client2;
   ARROW_CHECK_OK(client2.Connect("/tmp/store2", "/tmp/manager2",
                                  plasma::kPlasmaDefaultReleaseDelay));
-  ObjectID oid1 = ObjectID::from_random();
-  ObjectID oid2 = ObjectID::from_random();
+  ObjectID oid1 = random_object_id();
+  ObjectID oid2 = random_object_id();
 #define NUM_OBJ_REQUEST 2
 #define WAIT_TIMEOUT_MS 1000
   ObjectRequest obj_requests[NUM_OBJ_REQUEST];
@@ -236,8 +238,8 @@ TEST plasma_get_tests(void) {
                                  plasma::kPlasmaDefaultReleaseDelay));
   ARROW_CHECK_OK(client2.Connect("/tmp/store2", "/tmp/manager2",
                                  plasma::kPlasmaDefaultReleaseDelay));
-  ObjectID oid1 = ObjectID::from_random();
-  ObjectID oid2 = ObjectID::from_random();
+  ObjectID oid1 = random_object_id();
+  ObjectID oid2 = random_object_id();
   ObjectBuffer obj_buffer1;
 
   ObjectID oid_array1[1] = {oid1};
@@ -278,8 +280,8 @@ TEST plasma_get_multiple_tests(void) {
                                  plasma::kPlasmaDefaultReleaseDelay));
   ARROW_CHECK_OK(client2.Connect("/tmp/store2", "/tmp/manager2",
                                  plasma::kPlasmaDefaultReleaseDelay));
-  ObjectID oid1 = ObjectID::from_random();
-  ObjectID oid2 = ObjectID::from_random();
+  ObjectID oid1 = random_object_id();
+  ObjectID oid2 = random_object_id();
   ObjectID obj_ids[NUM_OBJ_REQUEST];
   ObjectBuffer obj_buffer[NUM_OBJ_REQUEST];
   int obj1_first = 1, obj2_first = 2;

src/plasma/test/manager_tests.cc

@@ -9,6 +9,8 @@
 
 #include <string>
 
+#include "plasma/test-util.h"
+
 #include "common.h"
 #include "test/test_common.h"
 #include "event_loop.h"
@@ -255,7 +257,7 @@ TEST object_notifications_test(void) {
   int flags = fcntl(fd[1], F_GETFL, 0);
   RAY_CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0);
 
-  ObjectID object_id = ObjectID::from_random();
+  ObjectID object_id = plasma::random_object_id();
   fb::ObjectInfoT info;
   info.object_id = object_id.binary();
   info.data_size = 10;

thirdparty/scripts/build_arrow.sh

@@ -34,10 +34,10 @@ else
   exit 1
 fi
 
-# The PR for this commit is https://github.com/apache/arrow/pull/2458. We
+# The PR for this commit is https://github.com/apache/arrow/pull/2482. 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=fda4b3dcfc773612b12973df5053193f236fc696
+TARGET_COMMIT_ID=927bd34aaad875e82beca2584d5d777839fa8bb0
 build_arrow() {
   echo "building arrow"
   # Make sure arrow will be built again when building ray for java later than python