class_docstr: State for a single active rechunk execution
  
  This object is responsible for splitting, sending, receiving and combining
  data shards.
  
  It is entirely agnostic to the distributed system and can perform a rechunk
  with other run instances using `rpc``.
  
  The user of this needs to guarantee that only `ArrayRechunkRun`s of the same unique
  `ShuffleID` and `run_id` interact.
  
  Parameters
  ----------
  worker_for:
      A mapping partition_id -> worker_address.
  old:
      Existing chunking of the array per dimension.
  new:
      Desired chunking of the array per dimension.
  id:
      A unique `ShuffleID` this belongs to.
  run_id:
      A unique identifier of the specific execution of the shuffle this belongs to.
  span_id:
      Span identifier; see :doc:`spans`
  local_address:
      The local address this Shuffle can be contacted by using `rpc`.
  directory:
      The scratch directory to buffer data in.
  executor:
      Thread pool to use for offloading compute.
  rpc:
      A callable returning a PooledRPCCall to contact other Shuffle instances.
      Typically a ConnectionPool.
  digest_metric:
      A callable to ingest a performance metric.
      Typically Server.digest_metric.
  scheduler:
      A PooledRPCCall to contact the scheduler.
  memory_limiter_disk:
  memory_limiter_comm:
      A ``ResourceLimiter`` limiting the total amount of memory used in either
      buffer.
class_docstr_tok: ["State","for","a","single","active","rechunk","execution","This","object","is","responsible","for","splitting",",","sending",",","receiving","and","combining","data","shards",".","It","is","entirely","agnostic","to","the","distributed","system","and","can","perform","a","rechunk","with","other","run","instances","using","`","rpc","``",".","The","user","of","this","needs","to","guarantee","that","only","`","ArrayRechunkRun","`","s","of","the","same","unique","`","ShuffleID","`","and","`","run_id","`","interact",".","Parameters","--","--","--","--","--","worker_for",":","A","mapping","partition_id","-",">","worker_address",".","old",":","Existing","chunking","of","the","array","per","dimension",".","new",":","Desired","chunking","of","the","array","per","dimension",".","id",":","A","unique","`","ShuffleID","`","this","belongs","to",".","run_id",":","A","unique","identifier","of","the","specific","execution","of","the","shuffle","this","belongs","to",".","span_id",":","Span","identifier",";","see",":","doc",":","`","spans","`","local_address",":","The","local","address","this","Shuffle","can","be","contacted","by","using","`","rpc","`",".","directory",":","The","scratch","directory","to","buffer","data","in",".","executor",":","Thread","pool","to","use","for","offloading","compute",".","rpc",":","A","callable","returning","a","PooledRPCCall","to","contact","other","Shuffle","instances",".","Typically","a","ConnectionPool",".","digest_metric",":","A","callable","to","ingest","a","performance","metric",".","Typically","Server.digest_metric",".","scheduler",":","A","PooledRPCCall","to","contact","the","scheduler",".","memory_limiter_disk",":","memory_limiter_comm",":","A","``","ResourceLimiter","``","limiting","the","total","amount","of","memory","used","in","either","buffer","."]
code: def _get_output_partition(
      self, partition_id: NDIndex, key: Key, **kwargs: Any
  ) -> np.ndarray:
      # Quickly read metadata from disk.
      # This is a bunch of seek()'s interleaved with short reads.
      data = self._read_from_disk(partition_id)
      # Copy the memory-mapped buffers from disk into memory.
      # This is where we'll spend most time.
      return convert_chunk(data)
  
imports: from __future__ import annotations
  import mmap
  import os
  from collections import defaultdict
  from collections.abc import Callable, Generator, Hashable, Sequence
  from concurrent.futures import ThreadPoolExecutor
  from dataclasses import dataclass
  from itertools import product
  from pathlib import Path
  from typing import TYPE_CHECKING, Any, NamedTuple
  import toolz
  from tornado.ioloop import IOLoop
  import dask
  from dask.base import tokenize
  from dask.highlevelgraph import HighLevelGraph, MaterializedLayer
  from dask.typing import Key
  from distributed.core import PooledRPCCall
  from distributed.metrics import context_meter
  from distributed.shuffle._core import NDIndex, ShuffleId, ShuffleRun, ShuffleSpec, get_worker_plugin, handle_transfer_errors, handle_unpack_errors
  from distributed.shuffle._limiter import ResourceLimiter
  from distributed.shuffle._pickle import unpickle_bytestream
  from distributed.shuffle._shuffle import barrier_key, shuffle_barrier
  from distributed.shuffle._worker_plugin import ShuffleWorkerPlugin
  from distributed.sizeof import sizeof
label_desc: Use image node_id 3 for calling the ArrayRechunkRun obj's underlying member method code with example usage: obj._get_output_partition(partition_id, key) and returns: convert_chunk