adding get callable

Author: fschlimb

Partitioned.md

@@ -14,9 +14,12 @@ While the interface is designed to be generic enough to cover any distribution b
 A conforming implementation of the partitioned-interface standard must provide and support a data structure object having a `__partitioned_interface__` method which returns a Python dictionary with the following fields:
 * `shape`: a tuple defining the number of partitions per dimension of the container's global data-(index-)space.
 * `partitions`: a dictionary mapping a position in the partition grid (as defined by `shape`) to a dictionary providing the partition object, the partition shape and locality information.
-* `locals`: Only for SPMD/MPI-like: list of the positions of the locally owned partitions. The positions serve as lookup keys in the `partitions` dictionary. Must not be available if not SPMD/MPI-like.
+* `locals`: Only for SPMD/MPI-like: list of the positions of the locally owned partitions. The positions serve as look-up keys in the `partitions` dictionary. Must not be available if not SPMD/MPI-like.
+* `get`: A callable converting a handle into a data object.
 
-In addition to the above required keys a container is encouraged to provide more information that could be potentially benefitial for consuming the distributed data structure. 
+In addition to the above
+* required keys a container is encouraged to provide more information that could be potentially beneficial for consuming the distributed data structure.
+* the dictionary must be pickle'able
 
 ## `shape`
 The shape of the partition grid must be of the same dimensionality as the underlying data-structure. `shape` provides the number of partitions along each dimension. Specifying `1` in a given dimension means the dimension is not cut.
@@ -47,78 +50,86 @@ In addition to the above required keys a container is encouraged to provide more
     * When the underlying backend supports it and for all non-SPMD backends partitions must be provided as references. This avoids unnecessary data movement.
       * Ray: ray.ObjectRef
       * Dask: dask.Future
+    * It is recommended to access the actual data through the callable in the 'get' field of `__partitioned__` differentiating between different handle types can be avoided and type checks can be limited to basic types like pandas.DataFrame and numpy.ndarray.
+
   * For SPMD-MPI-like backends: partitions which are not locally available may be `None`. This is the recommended behavior unless the underlying backend supports references such as promises to avoid unnecessary data movement.
 * `location`
   * The location information must include all necessary data to uniquely identify the location of the partition/data. The exact information depends on the underlying distribution system:
     * Ray: ip-address
     * Dask: worker-Id (name, ip, or ip:port)
-    * SPMD/MPI-like frameworks such as MPI, SHMEM etc: rank
+    * SPMD/MPI-like frameworks such as MPI, SHMEM etc.: rank
+
+## `get`
+This provides a callable which returns raw data object when called with a handle provided in the `data` field of an entry in `partition`. Raw data objects are standard data structures like pandas.DataFrame and numpy.ndarray.
 
 ## `locals`
 This is basically a short-cut for SPMD environments which allows processes/ranks to quickly extract the local partition. It saves processes from parsing the `partitions` dictionary for the local rank/address which is helpful when the number of ranks/processes/PEs is large.
 
+
 ## Examples
 ### 1d-data-structure (64 elements), 1d-partition-grid, 4 partitions on 4 nodes, blocked distribution, partitions are of type `Ray.ObjRef`, Ray
 ```python
 __partitioned_interface__ = {
-  ‘shape’: (4,),
-  ‘partitions’: {
+  'shape': (4,),
+  'partitions': {
       (0,):  {
         'start': (0,),
         'shape': (16,),
         'data': ObjRef0,
-        'location': ‘1.1.1.1’, }
+        'location': '1.1.1.1’, }
       (1,): {
         'start': (16,),
         'shape': (16,),
         'data': ObjRef1,
-        'location': ‘1.1.1.2’, }
+        'location': '1.1.1.2’, }
       (2,): {
         'start': (32,),
         'shape': (16,),
         'data': ObjRef2,
-        'location': ‘1.1.1.3’, }
+        'location': '1.1.1.3’, }
       (3,): {
         'start': (48,),
         'shape': (16,),
         'data': ObjRef3,
-        'location': ‘1.1.1.4’, }
-  }
+        'location': '1.1.1.4’, }
+  },
+  'get': lambda x: ray.get(x)
 }
 ```
 ### 2d-structure (64 elements), 2d-partition-grid, 4 partitions on 2 nodes, block-cyclic distribution, partitions are of type `dask.Future`, dask
 ```python
 __partitioned_interface__ = {
-  ‘shape’: (2,2),
-  ‘partitions’: {
+  'shape’: (2,2),
+  'partitions’: {
       (1,1): {
         'start': (4, 4),
         'shape': (4, 4),
         'data': future0,
-        'location': ‘Alice’, },
+        'location': 'Alice’, },
       (1,0):  {
         'start': (4, 0),
         'shape': (4, 4),
         'data': future1,
-        'location': ‘1.1.1.2:55667’, },
+        'location': '1.1.1.2:55667’, },
       (0,1):  {
         'start': (0, 4),
         'shape': (4, 4),
         'data': future2,
-        'location': ‘Alice’, },
+        'location': 'Alice’, },
       (0,0): {
         'start': (0,0),
         'shape': (4, 4),
         'data': future3,
-        'location': ‘1.1.1.2:55667’, },
+        'location': '1.1.1.2:55667’, },
   }
+  'get': lambda x: x.result()
 }
 ```
 ### 2d-structure (64 elements), 1d-partition-grid, 4 partitions on 2 ranks, row-block-cyclic distribution, partitions are of type `pandas.DataFrame`, MPI
 ```python
 __partitioned_interface__ = {
-  ‘shape’: (4,1),
-  ‘partitions’: {
+  'shape’: (4,1),
+  'partitions’: {
       (0,0):  {
         'start': (0, 0),
         'shape': (2, 8),
@@ -139,7 +150,8 @@ __partitioned_interface__ = {
         'shape': (2, 8),
         'data': None,   # this is for rank 0, for rank 1 it'd be df3
         'location': 1, },
-  }
+  },
+  'get': lambda x: x,
   'locals': [(0,0), (2,0)] # this is for rank 0, for rank 1 it'd be [(1,0), (3,0)]
 }
 ```