Delete the ObjectId.java and ObjectBuffer.java, change the input and output of plasma java client api from custem type to byte[]

Author: salah-man

java/plasma/src/main/java/org/apache/arrow/plasma/ObjectBuffer.java

@@ -17,7 +17,6 @@
  */
 package org.apache.arrow.plasma;
 
-import java.util.Collections;
 import java.util.List;
 
 /**
@@ -32,7 +31,7 @@ public interface ObjectStoreLink {
    * @param value The value to put in the object store.
    * @param metadata encodes whatever metadata the user wishes to encode.
    */
-  void put(ObjectId objectId, byte[] value, byte[] metadata);
+  void put(byte[] objectId, byte[] value, byte[] metadata);
 
   /**
    * Create a buffer from the PlasmaStore based on the <tt>objectId</tt>.
@@ -43,8 +42,9 @@ public interface ObjectStoreLink {
    * @param isMetadata false if get data, otherwise get metadata.
    * @return A PlasmaBuffer wrapping the object.
    */
-  default ObjectBuffer get(ObjectId objectId, int timeoutMs, boolean isMetadata) {
-    return get(Collections.singletonList(objectId), timeoutMs, isMetadata).get(0);
+  default byte[] get(byte[] objectId, int timeoutMs, boolean isMetadata) {
+    byte[][] objectIds = {objectId};
+    return get(objectIds, timeoutMs, isMetadata).get(0);
   }
 
   /**
@@ -56,7 +56,7 @@ default ObjectBuffer get(ObjectId objectId, int timeoutMs, boolean isMetadata) {
    * @param isMetadata false if get data, otherwise get metadata.
    * @return List of PlasmaBuffers wrapping objects.
    */
-  List<ObjectBuffer> get(List<? extends ObjectId> objectIds, int timeoutMs, boolean isMetadata);
+  List<byte[]> get(byte[][] objectIds, int timeoutMs, boolean isMetadata);
 
   /**
    * Wait until <tt>numReturns</tt> objects in <tt>objectIds</tt> are ready.
@@ -66,7 +66,7 @@ default ObjectBuffer get(ObjectId objectId, int timeoutMs, boolean isMetadata) {
    * @param numReturns We are waiting for this number of objects to be ready.
    * @return List of object IDs that are ready
    */
-  List<ObjectId> wait(List<? extends ObjectId> objectIds, int timeoutMs, int numReturns);
+  List<byte[]> wait(byte[][] objectIds, int timeoutMs, int numReturns);
 
   /**
    * Compute the hash of an object in the object store.
@@ -75,23 +75,24 @@ default ObjectBuffer get(ObjectId objectId, int timeoutMs, boolean isMetadata) {
    * @return A digest byte array contains object's SHA256 hash. <tt>null</tt> means that the object
    * isn't in the object store.
    */
-  byte[] hash(ObjectId objectId);
+  byte[] hash(byte[] objectId);
 
   /**
    * Fetch the object with the given ID from other plasma manager instances.
    *
    * @param objectId The object ID used to identify the object.
    */
-  default void fetch(ObjectId objectId) {
-    fetch(Collections.singletonList(objectId));
+  default void fetch(byte[] objectId) {
+    byte[][] objectIds = {objectId};
+    fetch(objectIds);
   }
 
   /**
    * Fetch the objects with the given IDs from other plasma manager instances.
    *
    * @param objectIds List of object IDs used to identify the objects.
    */
-  void fetch(List<? extends ObjectId> objectIds);
+  void fetch(byte[][] objectIds);
 
   /**
    * Evict some objects to recover given count of bytes.
@@ -100,4 +101,11 @@ default void fetch(ObjectId objectId) {
    * @return The number of bytes that have been evicted.
    */
   long evict(long numBytes);
+
+  /**
+   * Release the reference of the object.
+   *
+   * @param objectId The object ID used to release the reference of the object.
+   */
+  void release(byte[] objectId);
 }

java/plasma/src/main/java/org/apache/arrow/plasma/ObjectId.java

@@ -48,10 +48,10 @@ public PlasmaClient(String configFile, String configOverwrites, String storeSock
   // interface methods --------------------
 
   @Override
-  public void put(ObjectId objectId, byte[] value, byte[] metadata) {
+  public void put(byte[] objectId, byte[] value, byte[] metadata) {
     ByteBuffer buf = null;
     try {
-      buf = PlasmaClientJNI.create(conn, objectId.getBytes(), value.length, metadata);
+      buf = PlasmaClientJNI.create(conn, objectId, value.length, metadata);
     } catch (Exception e) {
       System.err.println("ObjectId " + objectId + " error at PlasmaClient put");
       e.printStackTrace();
@@ -61,49 +61,37 @@ public void put(ObjectId objectId, byte[] value, byte[] metadata) {
     }
 
     buf.put(value);
-    PlasmaClientJNI.seal(conn, objectId.getBytes());
-    PlasmaClientJNI.release(conn, objectId.getBytes());
+    PlasmaClientJNI.seal(conn, objectId);
+    PlasmaClientJNI.release(conn, objectId);
   }
 
   @Override
-  public List<ObjectBuffer> get(List<? extends ObjectId> objectIds, int timeoutMs, boolean isMetadata) {
-    byte[][] ids = getIdBytes(objectIds);
-    ByteBuffer[][] bufs = PlasmaClientJNI.get(conn, ids, timeoutMs);
-    assert bufs.length == objectIds.size();
+  public List<byte[]> get(byte[][] objectIds, int timeoutMs, boolean isMetadata) {
+    ByteBuffer[][] bufs = PlasmaClientJNI.get(conn, objectIds, timeoutMs);
+    assert bufs.length == objectIds.length;
 
-    List<ObjectBuffer> ret = new ArrayList<>();
+    List<byte[]> ret = new ArrayList<>();
     for (int i = 0; i < bufs.length; i++) {
-      ObjectId oid = objectIds.get(i);
       ByteBuffer buf = bufs[i][isMetadata ? 1 : 0];
       if (buf == null) {
-        ret.add(new ObjectBuffer(null, null));
+        ret.add(null);
       } else {
         byte[] bb = new byte[buf.remaining()];
         buf.get(bb);
-        ret.add(new ObjectBuffer(bb, (byte[] b) -> this.release(oid)));
+        ret.add(bb);
       }
     }
     return ret;
   }
 
-  private static byte[][] getIdBytes(List<? extends ObjectId> objectIds) {
-    int size = objectIds.size();
-    byte[][] ids = new byte[size][];
-    for (int i = 0; i < size; i++) {
-      ids[i] = objectIds.get(i).getBytes();
-    }
-    return ids;
-  }
-
   @Override
-  public List<ObjectId> wait(List<? extends ObjectId> objectIds, int timeoutMs, int numReturns) {
-    byte[][] ids = getIdBytes(objectIds);
-    byte[][] readys = PlasmaClientJNI.wait(conn, ids, timeoutMs, numReturns);
+  public List<byte[]> wait(byte[][] objectIds, int timeoutMs, int numReturns) {
+    byte[][] readys = PlasmaClientJNI.wait(conn, objectIds, timeoutMs, numReturns);
 
-    List<ObjectId> ret = new ArrayList<>();
+    List<byte[]> ret = new ArrayList<>();
     for (byte[] ready : readys) {
-      for (ObjectId id : objectIds) {
-        if (Arrays.equals(ready, id.getBytes())) {
+      for (byte[] id : objectIds) {
+        if (Arrays.equals(ready, id)) {
           ret.add(id);
           break;
         }
@@ -115,14 +103,13 @@ public List<ObjectId> wait(List<? extends ObjectId> objectIds, int timeoutMs, in
   }
 
   @Override
-  public byte[] hash(ObjectId objectId) {
-    return PlasmaClientJNI.hash(conn, objectId.getBytes());
+  public byte[] hash(byte[] objectId) {
+    return PlasmaClientJNI.hash(conn, objectId);
   }
 
   @Override
-  public void fetch(List<? extends ObjectId> objectIds) {
-    byte[][] ids = getIdBytes(objectIds);
-    PlasmaClientJNI.fetch(conn, ids);
+  public void fetch(byte[][] objectIds) {
+    PlasmaClientJNI.fetch(conn, objectIds);
   }
 
   @Override
@@ -138,25 +125,25 @@ public long evict(long numBytes) {
    *
    * @param objectId used to identify an object.
    */
-  public void seal(ObjectId objectId) {
-    PlasmaClientJNI.seal(conn, objectId.getBytes());
+  public void seal(byte[] objectId) {
+    PlasmaClientJNI.seal(conn, objectId);
   }
 
   /**
    * Notify Plasma that the object is no longer needed.
    *
    * @param objectId used to identify an object.
    */
-  public void release(ObjectId objectId) {
-    PlasmaClientJNI.release(conn, objectId.getBytes());
+  public void release(byte[] objectId) {
+    PlasmaClientJNI.release(conn, objectId);
   }
 
   /**
    * Check if the object is present and has been sealed in the PlasmaStore.
    *
    * @param objectId used to identify an object.
    */
-  public boolean contains(ObjectId objectId) {
-    return PlasmaClientJNI.contains(conn, objectId.getBytes());
+  public boolean contains(byte[] objectId) {
+    return PlasmaClientJNI.contains(conn, objectId);
   }
 }