code: def from_partition_refs(ip2part_id, part_id2ref, old_rdd):
      ray_ctx = RayContext.get()
      uuid_str = str(uuid.uuid4())
      id2store_name = {}
      partition_stores = {}
      part_id2ip = {}
      result = []
      for node, part_ids in ip2part_id.items():
          name = f"partition:{uuid_str}:{node}"
          store = (
              ray.remote(num_cpus=0, resources={f"node:{node}": 1e-4})(
                  LocalStore
              )
              .options(name=name)
              .remote()
          )
          partition_stores[name] = store
          for idx in part_ids:
              result.append(
                  store.upload_partition.remote(idx, part_id2ref[idx])
              )
              id2store_name[idx] = name
              part_id2ip[idx] = node
      ray.get(result)
      new_id_ip_store_rdd = old_rdd.mapPartitionsWithIndex(
          lambda idx, _: [(idx, part_id2ip[idx], id2store_name[idx])]
      ).cache()
      return RayXShards(uuid_str, new_id_ip_store_rdd, partition_stores)
  
imports: from collections import defaultdict
  import ray
  import ray._private.services
  import uuid
  import random
  from packaging import version
  from zoo.orca.data import XShards
  from zoo.ray import RayContext
  import logging
label_desc: Use image node_id 12 for calling the RayXShards obj's underlying member method code with example usage: obj.from_partition_refs(ip2part_id, part_id2ref, old_rdd) and returns: RayXShards