code: def from_partition_refs(
      ip2part_id: DefaultDict[str, List[int]],
      part_id2ref: Dict[int, "ObjectRef"],
      old_rdd: "RDD[Any]",
  ) -> "RayXShards":
      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 __future__ import annotations
  from collections import defaultdict
  import ray
  import ray._private.services
  import uuid
  import random
  from bigdl.orca.data import XShards
  from bigdl.orca.ray import OrcaRayContext
  import logging
  from bigdl.dllib.utils.log4Error import invalidInputError
  from numpy import ndarray
  from typing import TYPE_CHECKING, Any, Callable, DefaultDict, Dict, List, Tuple, Union, no_type_check
label_desc: Use image node_id 14 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