Repository: activeloopai/deeplake
Pull request: Improved composite transforms, added lazy mode toggle, added .compute() to Dataset and DatasetView
Improved composite transforms, added lazy mode toggle, added .compute() to Dataset and DatasetView.
Also refactored some old code in Dataset, DatasetView, and Tensorview. 
Also, fixes transforms breaking from loop when some empty samples are returned.

New implementation of transform is more efficient for multiple transforms as it doesn't store intermediate transformations however we lose the slicing logic that we usually have for Dataset and DatasetView. Another change is that .compute() doesn't need to be called within transform.

Might need to add more tests for transforms before merging.


Problem statement:
Uploading text dataset 
I tried uploading text files to Hub using this code: 

```
import numpy as np
from glob import glob
import time

import hub
from hub.schema import Text
from hub import schema, transform

my_schema = {
        "stocks": Text(shape=(None,))
}

fnames = glob("/home/hemant/Downloads/Data/Stocks/*")

@transform(schema=my_schema)
def load_data(sample):
    with open(sample) as f:
        text = f.readlines()
    return{
        "stocks": text
}

stime = time.time()
ds = load_data(fnames)
etime = time.time()

print("Time: ", etime - stime)

tag = "hemantr05/Stocks"
ds2 = ds.store(tag)
```

And got the following error:

Time:  9.655952453613281e-05
Traceback (most recent call last):
  File "hub-stock-data.py", line 33, in <module>
    ds2 = ds.store(tag)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/hub/compute/transform.py", line 342, in store
    n_samples = get_sample_size(self.schema, self.workers)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/hub/compute/transform.py", line 40, in get_sample_size
    samples = min(samples, (16 * 1024 * 1024 * 8) // (prod(shp) * sz))
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/hub/compute/transform.py", line 37, in prod
    res *= s
TypeError: unsupported operand type(s) for *=: 'int' and 'NoneType'

How can I fix it?

Hints from issue discussion:
@Hemantr05 we just fixed a couple of transform-related issues. Could you please kindly test if the same issue persists in the `fixes/stacked_transform` branch?  
@Hemantr05 I believe the issue here is that max_shape hasn't been set for the Text. max_shape is the max number of characters across all files that you expect. @edogrigqv2 this should have raised an appropriate error I think.

try this:-
```python
my_schema = {
        "stocks": Text(shape=(None,), max_shape=(1000,))
}
```
You can change 1000 if your text files are larger.
> @Hemantr05 we just fixed a couple of transform-related issues. Could you please kindly test if the same issue persists in the `fixes/stacked_transform` branch?

@davidbuniat I get the same issue
@AbhinavTuli after making the changes I get this error:

Traceback (most recent call last):
  File "/home/hemant/Desktop/Github/hub/Hub/hub/api/dataset.py", line 170, in __init__
    self.meta = self._store_meta()
  File "/home/hemant/Desktop/Github/hub/Hub/hub/api/dataset.py", line 231, in _store_meta
    self._fs_map["meta.json"] = bytes(json.dumps(meta), "utf-8")
  File "/home/hemant/Desktop/Github/hub/Hub/hub/store/store.py", line 144, in __setitem__
    self._map[slice_] = value
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/fsspec/mapping.py", line 154, in __setitem__
    self.fs.pipe_file(key, value)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/fsspec/spec.py", line 631, in pipe_file
    f.write(value)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/fsspec/spec.py", line 1516, in __exit__
    self.close()
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/fsspec/spec.py", line 1483, in close
    self.flush(force=True)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/fsspec/spec.py", line 1356, in flush
    if self._upload_chunk(final=force) is not False:
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/s3fs/core.py", line 1252, in _upload_chunk
    self.commit()
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/s3fs/core.py", line 1269, in commit
    Key=self.key, Bucket=self.bucket, Body=data, **self.kwargs
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/s3fs/core.py", line 1131, in _call_s3
    **kwargs)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/s3fs/core.py", line 200, in _call_s3
    return method(**additional_kwargs)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/botocore/client.py", line 357, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/botocore/client.py", line 676, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidAccessKeyId) when calling the PutObject operation: The AWS Access Key Id you provided does not exist in our records.

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "example1.py", line 29, in <module>
    ds2 = ds.store(tag)
  File "/home/hemant/Desktop/Github/hub/Hub/hub/compute/transform.py", line 432, in store
    ds_out = self.create_dataset(url, length=length, token=token, public=public)
  File "/home/hemant/Desktop/Github/hub/Hub/hub/compute/transform.py", line 253, in create_dataset
    public=public,
  File "/home/hemant/Desktop/Github/hub/Hub/hub/api/dataset.py", line 179, in __init__
    self._fs.rm(self._path, recursive=True)
  File "/home/hemant/anaconda3/lib/python3.7/site-packages/s3fs/core.py", line 994, in rm
    raise FileNotFoundError(path)
FileNotFoundError: s3://snark-hub/public/hemantr05/Stocks

Commit message: linting fix

Patch to review:
diff --git a/hub/api/dataset.py b/hub/api/dataset.py
index 1cfda4f778..1536c8dcd5 100644
--- a/hub/api/dataset.py
+++ b/hub/api/dataset.py
@@ -21,7 +21,12 @@
 from hub.log import logger
 from hub.api.tensorview import TensorView
 from hub.api.datasetview import DatasetView
-from hub.api.dataset_utils import slice_extract_info, slice_split, str_to_int
+from hub.api.dataset_utils import (
+    create_numpy_dict,
+    slice_extract_info,
+    slice_split,
+    str_to_int,
+)
 
 import hub.schema.serialize
 import hub.schema.deserialize
@@ -67,7 +72,8 @@ def __init__(
         storage_cache: int = defaults.DEFAULT_STORAGE_CACHE_SIZE,
         lock_cache=True,
         tokenizer=None,
-        public=True,
+        lazy: bool = True,
+        public: bool = True,
     ):
         """| Open a new or existing dataset for read/write
 
@@ -95,6 +101,8 @@ def __init__(
             if 0, False or None, then storage cache is not used
         lock_cache: bool, optional
             Lock the cache for avoiding multiprocessing errors
+        lazy: bool, optional
+            Setting this to False will stop lazy computation and will allow items to be accessed without .compute()
         public: bool, optional
             only applicable if using hub storage, ignored otherwise
             setting this to False allows only the user who created it to access the dataset and
@@ -113,6 +121,7 @@ def __init__(
         self._token = token
         self._mode = mode
         self.tokenizer = tokenizer
+        self.lazy = lazy
 
         self._fs, self._path = (
             (fs, url) if fs else get_fs_and_path(self._url, token=token, public=public)
@@ -351,17 +360,31 @@ def __getitem__(self, slice_):
                 num_samples=num,
                 offset=ofs,
                 squeeze_dim=isinstance(slice_list[0], int),
+                lazy=self.lazy,
             )
         elif not slice_list:
             if subpath in self._tensors.keys():
-                return TensorView(
-                    dataset=self, subpath=subpath, slice_=slice(0, self.shape[0])
+                tensorview = TensorView(
+                    dataset=self,
+                    subpath=subpath,
+                    slice_=slice(0, self.shape[0]),
+                    lazy=self.lazy,
                 )
+                if self.lazy:
+                    return tensorview
+                else:
+                    return tensorview.compute()
             return self._get_dictionary(subpath)
         else:
             num, ofs = slice_extract_info(slice_list[0], self.shape[0])
             if subpath in self._tensors.keys():
-                return TensorView(dataset=self, subpath=subpath, slice_=slice_list)
+                tensorview = TensorView(
+                    dataset=self, subpath=subpath, slice_=slice_list, lazy=self.lazy
+                )
+                if self.lazy:
+                    return tensorview
+                else:
+                    return tensorview.compute()
             if len(slice_list) > 1:
                 raise ValueError("You can't slice a dictionary of Tensors")
             return self._get_dictionary(subpath, slice_list[0])
@@ -422,7 +445,7 @@ def delete(self):
 
     def to_pytorch(
         self,
-        Transform=None,
+        transform=None,
         inplace=True,
         output_type=dict,
         offset=None,
@@ -432,7 +455,7 @@ def to_pytorch(
 
         Parameters
         ----------
-        Transform: function that transforms data in a dict format
+        transform: function that transforms data in a dict format
         inplace: bool, optional
             Defines if data should be converted to torch.Tensor before or after Transforms applied (depends on what data
             type you need for Transforms). Default is True.
@@ -445,15 +468,14 @@ def to_pytorch(
         """
         if "torch" not in sys.modules:
             raise ModuleNotInstalledException("torch")
-        else:
-            import torch
+        import torch
 
-            global torch
+        global torch
 
         self.flush()  # FIXME Without this some tests in test_converters.py fails, not clear why
         return TorchDataset(
             self,
-            Transform,
+            transform,
             inplace=inplace,
             output_type=output_type,
             offset=offset,
@@ -548,11 +570,12 @@ def _get_dictionary(self, subpath, slice_=None):
                     if split_key[i] not in cur.keys():
                         cur[split_key[i]] = {}
                     cur = cur[split_key[i]]
-                slice_ = slice_ if slice_ else slice(0, self.shape[0])
-                cur[split_key[-1]] = TensorView(
-                    dataset=self, subpath=key, slice_=slice_
+                slice_ = slice_ or slice(0, self.shape[0])
+                tensorview = TensorView(
+                    dataset=self, subpath=key, slice_=slice_, lazy=self.lazy
                 )
-        if len(tensor_dict) == 0:
+                cur[split_key[-1]] = tensorview if self.lazy else tensorview.compute()
+        if not tensor_dict:
             raise KeyError(f"Key {subpath} was not found in dataset")
         return tensor_dict
 
@@ -565,6 +588,12 @@ def __len__(self):
         """ Number of samples in the dataset """
         return self.shape[0]
 
+    def disable_lazy(self):
+        self.lazy = False
+
+    def enable_lazy(self):
+        self.lazy = True
+
     def flush(self):
         """Save changes from cache to dataset final storage.
         Does not invalidate this object.
@@ -593,8 +622,14 @@ def _update_dataset_state(self):
                 self.username, self.dataset_name, "UPLOADED"
             )
 
+    def numpy(self):
+        return [create_numpy_dict(self, i) for i in range(self.shape[0])]
+
+    def compute(self):
+        return self.numpy()
+
     def __str__(self):
-        out = (
+        return (
             "Dataset(schema="
             + str(self.schema)
             + "url="
@@ -608,7 +643,6 @@ def __str__(self):
             + self.mode
             + "')"
         )
-        return out
 
     def __repr__(self):
         return self.__str__()
@@ -690,7 +724,7 @@ def transform_numpy(sample):
             for k, v in sample.items():
                 k = k.replace("/", "_")
                 if not isinstance(v, dict):
-                    if isinstance(v, tuple) or isinstance(v, list):
+                    if isinstance(v, (tuple, list)):
                         new_v = list(v)
                         for i in range(len(new_v)):
                             new_v[i] = new_v[i].numpy()
@@ -796,8 +830,7 @@ def dict_sampling(d, path=""):
 
         def generate_schema(ds):
             tf_schema = ds[1].features
-            schema = to_hub(tf_schema).dict_
-            return schema
+            return to_hub(tf_schema).dict_
 
         def to_hub(tf_dt, max_shape=None, path=""):
             if isinstance(tf_dt, tfds.features.FeaturesDict):
@@ -909,10 +942,7 @@ def transform_numpy(sample):
             d = {}
             for k, v in sample.items():
                 k = k.replace("/", "_")
-                if not isinstance(v, dict):
-                    d[k] = v.numpy()
-                else:
-                    d[k] = transform_numpy(v)
+                d[k] = transform_numpy(v) if isinstance(v, dict) else v.numpy()
             return d
 
         @hub.transform(schema=my_schema, scheduler=scheduler, workers=workers)
@@ -984,9 +1014,9 @@ def dict_to_hub(dic, path=""):
                     value_shape = v.shape if hasattr(v, "shape") else ()
                     if isinstance(v, torch.Tensor):
                         v = v.numpy()
-                    shape = tuple([None for it in value_shape])
+                    shape = tuple(None for it in value_shape)
                     max_shape = (
-                        max_dict[cur_path] or tuple([10000 for it in value_shape])
+                        max_dict[cur_path] or tuple(10000 for it in value_shape)
                         if not isinstance(v, str)
                         else (10000,)
                     )
@@ -1005,10 +1035,7 @@ def transform_numpy(sample):
             d = {}
             for k, v in sample.items():
                 k = k.replace("/", "_")
-                if not isinstance(v, dict):
-                    d[k] = v
-                else:
-                    d[k] = transform_numpy(v)
+                d[k] = transform_numpy(v) if isinstance(v, dict) else v
             return d
 
         @hub.transform(schema=my_schema, scheduler=scheduler, workers=workers)
@@ -1059,11 +1086,9 @@ def __getitem__(self, index):
             split_key = key.split("/")
             cur = d
             for i in range(1, len(split_key) - 1):
-                if split_key[i] in cur.keys():
-                    cur = cur[split_key[i]]
-                else:
+                if split_key[i] not in cur.keys():
                     cur[split_key[i]] = {}
-                    cur = cur[split_key[i]]
+                cur = cur[split_key[i]]
             if not isinstance(self._ds._tensors[key][index], bytes) and not isinstance(
                 self._ds._tensors[key][index], str
             ):
@@ -1078,23 +1103,5 @@ def __getitem__(self, index):
 
     def __iter__(self):
         self._init_ds()
-        start = self.offset if self.offset is not None else 0
-        for index in range(start, start + self.__len__()):
-            d = {}
-            for key in self._ds._tensors.keys():
-                split_key = key.split("/")
-                cur = d
-                for i in range(1, len(split_key) - 1):
-                    if split_key[i] in cur.keys():
-                        cur = cur[split_key[i]]
-                    else:
-                        cur[split_key[i]] = {}
-                        cur = cur[split_key[i]]
-                t = self._ds._tensors[key][index]
-                if self.inplace:
-                    t = torch.tensor(t)
-                cur[split_key[-1]] = t
-            d = self._do_transform(d)
-            if self.inplace & (self.output_type != dict) & (type(d) == dict):
-                d = self.output_type(d.values())
-            yield (d)
+        for i in range(len(self)):
+            yield self[i]
diff --git a/hub/api/dataset_utils.py b/hub/api/dataset_utils.py
index 209d682fbb..554a7d397e 100644
--- a/hub/api/dataset_utils.py
+++ b/hub/api/dataset_utils.py
@@ -10,7 +10,7 @@ def slice_split(slice_):
     for sl in slice_:
         if isinstance(sl, str):
             path += sl if sl.startswith("/") else "/" + sl
-        elif isinstance(sl, int) or isinstance(sl, slice):
+        elif isinstance(sl, (int, slice)):
             list_slice.append(sl)
         else:
             raise TypeError(
@@ -51,13 +51,26 @@ def slice_extract_info(slice_, num):
             )
     if slice_.start is not None and slice_.stop is not None:
         num = 0 if slice_.stop < slice_.start else slice_.stop - slice_.start
-    elif slice_.start is None and slice_.stop is not None:
+    elif slice_.stop is not None:
         num = slice_.stop
-    elif slice_.start is not None and slice_.stop is None:
+    elif slice_.start is not None:
         num = num - slice_.start
     return num, offset
 
 
+def create_numpy_dict(dataset, index):
+    numpy_dict = {}
+    for path in dataset._tensors.keys():
+        d = numpy_dict
+        split = path.split("/")
+        for subpath in split[1:-1]:
+            if subpath not in d:
+                d[subpath] = {}
+            d = d[subpath]
+        d[split[-1]] = dataset[path, index].numpy()
+    return numpy_dict
+
+
 def str_to_int(assign_value, tokenizer):
     if isinstance(assign_value, bytes):
         try:
@@ -73,10 +86,9 @@ def str_to_int(assign_value, tokenizer):
     if tokenizer is not None:
         if "transformers" not in sys.modules:
             raise ModuleNotInstalledException("transformers")
-        else:
-            import transformers
+        import transformers
 
-            global transformers
+        global transformers
         tokenizer = transformers.AutoTokenizer.from_pretrained("bert-base-cased")
         assign_value = (
             np.array(tokenizer(assign_value, add_special_tokens=False)["input_ids"])
diff --git a/hub/api/datasetview.py b/hub/api/datasetview.py
index 8e91663e71..473202027c 100644
--- a/hub/api/datasetview.py
+++ b/hub/api/datasetview.py
@@ -1,5 +1,10 @@
 from hub.api.tensorview import TensorView
-from hub.api.dataset_utils import slice_extract_info, slice_split, str_to_int
+from hub.api.dataset_utils import (
+    create_numpy_dict,
+    slice_extract_info,
+    slice_split,
+    str_to_int,
+)
 from hub.exceptions import NoneValueException
 import collections.abc as abc
 
@@ -8,9 +13,10 @@ class DatasetView:
     def __init__(
         self,
         dataset=None,
-        num_samples=None,
-        offset=None,
-        squeeze_dim=False,
+        num_samples: int = None,
+        offset: int = None,
+        squeeze_dim: bool = False,
+        lazy: bool = True,
     ):
         """Creates a DatasetView object for a subset of the Dataset.
 
@@ -22,8 +28,10 @@ def __init__(
             The number of samples in this DatasetView
         offset: int
             The offset from which the DatasetView starts
-        squeeze_dim: bool
+        squeeze_dim: bool, optional
             For slicing with integers we would love to remove the first dimension to make it nicer
+        lazy: bool, optional
+            Setting this to False will stop lazy computation and will allow items to be accessed without .compute()
         """
         if dataset is None:
             raise NoneValueException("dataset")
@@ -36,6 +44,7 @@ def __init__(
         self.num_samples = num_samples
         self.offset = offset
         self.squeeze_dim = squeeze_dim
+        self.lazy = lazy
 
     def __getitem__(self, slice_):
         """| Gets a slice or slices from DatasetView
@@ -63,31 +72,38 @@ def __getitem__(self, slice_):
                 num_samples=num,
                 offset=ofs + self.offset,
                 squeeze_dim=isinstance(slice_list[0], int),
+                lazy=self.lazy,
             )
         elif not slice_list:
-            slice_ = slice(self.offset, self.offset + self.num_samples)
+            slice_ = (
+                slice(self.offset, self.offset + self.num_samples)
+                if not self.squeeze_dim
+                else self.offset
+            )
             if subpath in self.dataset._tensors.keys():
-                return TensorView(
+                tensorview = TensorView(
                     dataset=self.dataset,
                     subpath=subpath,
                     slice_=slice_,
-                    squeeze_dims=[True] if self.squeeze_dim else [],
+                    lazy=self.lazy,
                 )
-            return self._get_dictionary(self.dataset, subpath, slice=slice_)
+                return tensorview if self.lazy else tensorview.compute()
+            return self._get_dictionary(subpath, slice=slice_)
         else:
             num, ofs = slice_extract_info(slice_list[0], self.num_samples)
             slice_list[0] = (
                 ofs + self.offset
-                if num == 1
+                if isinstance(slice_list[0], int)
                 else slice(ofs + self.offset, ofs + self.offset + num)
             )
             if subpath in self.dataset._tensors.keys():
-                return TensorView(
+                tensorview = TensorView(
                     dataset=self.dataset,
                     subpath=subpath,
                     slice_=slice_list,
-                    squeeze_dims=[True] if self.squeeze_dim else [],
+                    lazy=self.lazy,
                 )
+                return tensorview if self.lazy else tensorview.compute()
             if len(slice_list) > 1:
                 raise ValueError("You can't slice a dictionary of Tensors")
             return self._get_dictionary(subpath, slice_list[0])
@@ -110,12 +126,9 @@ def __setitem__(self, slice_, value):
         slice_list = [0] + slice_list if self.squeeze_dim else slice_list
         if not subpath:
             raise ValueError("Can't assign to dataset sliced without subpath")
-        elif not slice_list:
-            slice_ = (
-                self.offset
-                if self.num_samples == 1
-                else slice(self.offset, self.offset + self.num_samples)
-            )
+
+        if not slice_list:
+            slice_ = slice(self.offset, self.offset + self.num_samples)
             self.dataset._tensors[subpath][slice_] = assign_value  # Add path check
         else:
             num, ofs = (
@@ -125,7 +138,7 @@ def __setitem__(self, slice_, value):
             )
             slice_list[0] = (
                 slice(ofs + self.offset, ofs + self.offset + num)
-                if num > 1
+                if isinstance(slice_list[0], slice)
                 else ofs + self.offset
             )
             self.dataset._tensors[subpath][slice_list] = assign_value
@@ -137,7 +150,7 @@ def keys(self):
         """
         return self.dataset._tensors.keys()
 
-    def _get_dictionary(self, subpath, slice_=None):
+    def _get_dictionary(self, subpath, slice_):
         """Gets dictionary from dataset given incomplete subpath"""
         tensor_dict = {}
         subpath = subpath if subpath.endswith("/") else subpath + "/"
@@ -146,18 +159,18 @@ def _get_dictionary(self, subpath, slice_=None):
                 suffix_key = key[len(subpath) :]
                 split_key = suffix_key.split("/")
                 cur = tensor_dict
-                for i in range(len(split_key) - 1):
-                    if split_key[i] not in cur.keys():
-                        cur[split_key[i]] = {}
-                    cur = cur[split_key[i]]
-                slice_ = slice_ if slice_ else slice(0, self.dataset.shape[0])
-                cur[split_key[-1]] = TensorView(
+                for sub_key in split_key[:-1]:
+                    if sub_key not in cur.keys():
+                        cur[sub_key] = {}
+                    cur = cur[sub_key]
+                tensorview = TensorView(
                     dataset=self.dataset,
                     subpath=key,
                     slice_=slice_,
-                    squeeze_dims=[True] if self.squeeze_dim else [],
+                    lazy=self.lazy,
                 )
-        if len(tensor_dict) == 0:
+                cur[split_key[-1]] = tensorview if self.lazy else tensorview.compute()
+        if not tensor_dict:
             raise KeyError(f"Key {subpath} was not found in dataset")
         return tensor_dict
 
@@ -193,10 +206,19 @@ def to_tensorflow(self):
             num_samples=self.num_samples, offset=self.offset
         )
 
-    def to_pytorch(self, Transform=None):
+    def to_pytorch(
+        self,
+        transform=None,
+        inplace=True,
+        output_type=dict,
+    ):
         """Converts the dataset into a pytorch compatible format"""
         return self.dataset.to_pytorch(
-            Transform=Transform, num_samples=self.num_samples, offset=self.offset
+            transform=transform,
+            num_samples=self.num_samples,
+            offset=self.offset,
+            inplace=inplace,
+            output_type=output_type,
         )
 
     def resize_shape(self, size: int) -> None:
@@ -206,3 +228,21 @@ def resize_shape(self, size: int) -> None:
     def commit(self) -> None:
         """Commit dataset"""
         self.dataset.commit()
+
+    def numpy(self):
+        if self.num_samples == 1 and self.squeeze_dim:
+            return create_numpy_dict(self.dataset, self.offset)
+        else:
+            return [
+                create_numpy_dict(self.dataset, self.offset + i)
+                for i in range(self.num_samples)
+            ]
+
+    def disable_lazy(self):
+        self.lazy = False
+
+    def enable_lazy(self):
+        self.lazy = True
+
+    def compute(self):
+        return self.numpy()
diff --git a/hub/api/tensorview.py b/hub/api/tensorview.py
index 3a13b7dae8..d36d157b28 100644
--- a/hub/api/tensorview.py
+++ b/hub/api/tensorview.py
@@ -10,7 +10,7 @@ def __init__(
         dataset=None,
         subpath=None,
         slice_=None,
-        squeeze_dims=[],
+        lazy: bool = True,
     ):
         """Creates a TensorView object for a particular tensor in the dataset
 
@@ -22,6 +22,8 @@ def __init__(
             The full path to the particular Tensor in the Dataset
         slice_: optional
             The `slice_` of this Tensor that needs to be accessed
+        lazy: bool, optional
+            Setting this to False will stop lazy computation and will allow items to be accessed without .compute()
         """
 
         if dataset is None:
@@ -31,10 +33,11 @@ def __init__(
 
         self.dataset = dataset
         self.subpath = subpath
+        self.lazy = lazy
 
-        if isinstance(slice_, int) or isinstance(slice_, slice):
+        if isinstance(slice_, (int, slice)):
             self.slice_ = [slice_]
-        elif isinstance(slice_, tuple) or isinstance(slice_, list):
+        elif isinstance(slice_, (tuple, list)):
             self.slice_ = list(slice_)
         self.nums = []
         self.offsets = []
@@ -46,7 +49,7 @@ def __init__(
                 self.offsets.append(it)
                 self.squeeze_dims.append(True)
             elif isinstance(it, slice):
-                ofs = it.start if it.start else 0
+                ofs = it.start or 0
                 num = it.stop - ofs if it.stop else None
                 self.nums.append(num)
                 self.offsets.append(ofs)
@@ -69,9 +72,9 @@ def numpy(self):
                 tokenizer = AutoTokenizer.from_pretrained("bert-base-cased")
                 if value.ndim == 1:
                     return tokenizer.decode(value.tolist())
-            else:
-                if value.ndim == 1:
-                    return "".join(chr(it) for it in value.tolist())
+            elif value.ndim == 1:
+                return "".join(chr(it) for it in value.tolist())
+            raise ValueError("Can only access Text with integer index")
         return self.dataset._tensors[self.subpath][self.slice_]
 
     def compute(self):
@@ -93,26 +96,28 @@ def __getitem__(self, slice_):
 
         if subpath:
             raise ValueError("Can't slice a Tensor with string")
-        else:
-            new_nums = self.nums.copy()
-            new_offsets = self.offsets.copy()
-            if len(new_nums) < len(slice_list):
-                new_nums.extend([None] * (len(slice_list) - len(new_nums)))
-                new_offsets.extend([0] * (len(slice_list) - len(new_offsets)))
-            for i in range(len(slice_list)):
-                slice_list[i] = self._combine(
-                    slice_list[i], new_nums[i], new_offsets[i]
-                )
-            for i in range(len(slice_list), len(new_nums)):
-                cur_slice = (
-                    slice(new_offsets[i], new_offsets[i] + new_nums[i])
-                    if new_nums[i] > 1
-                    else new_offsets[i]
-                )
-                slice_list.append(cur_slice)
-            return TensorView(
-                dataset=self.dataset, subpath=self.subpath, slice_=slice_list
+        new_nums = self.nums.copy()
+        new_offsets = self.offsets.copy()
+        if len(new_nums) < len(slice_list):
+            new_nums.extend([None] * (len(slice_list) - len(new_nums)))
+            new_offsets.extend([0] * (len(slice_list) - len(new_offsets)))
+        for i in range(len(slice_list)):
+            slice_list[i] = self._combine(slice_list[i], new_nums[i], new_offsets[i])
+        for i in range(len(slice_list), len(new_nums)):
+            cur_slice = (
+                slice(new_offsets[i], new_offsets[i] + new_nums[i])
+                if new_nums[i] > 1
+                else new_offsets[i]
             )
+            slice_list.append(cur_slice)
+        tensorview = TensorView(
+            dataset=self.dataset,
+            subpath=self.subpath,
+            slice_=slice_list,
+            lazy=self.lazy,
+        )
+
+        return tensorview if self.lazy else tensorview.compute()
 
     def __setitem__(self, slice_, value):
         """| Sets a slice or slices with a value
@@ -135,22 +140,21 @@ def __setitem__(self, slice_, value):
             raise ValueError(
                 "Can't slice a Tensor with multiple slices without subpath"
             )
-        else:
-            new_nums = self.nums.copy()
-            new_offsets = self.offsets.copy()
-            if len(new_nums) < len(slice_list):
-                new_nums.extend([None] * (len(slice_list) - len(new_nums)))
-                new_offsets.extend([0] * (len(slice_list) - len(new_offsets)))
-            for i in range(len(slice_list)):
-                slice_list[i] = self._combine(slice_[i], new_nums[i], new_offsets[i])
-            for i in range(len(slice_list), len(new_nums)):
-                cur_slice = (
-                    slice(new_offsets[i], new_offsets[i] + new_nums[i])
-                    if new_nums[i] > 1
-                    else new_offsets[i]
-                )
-                slice_list.append(cur_slice)
-            self.dataset._tensors[self.subpath][slice_list] = assign_value
+        new_nums = self.nums.copy()
+        new_offsets = self.offsets.copy()
+        if len(new_nums) < len(slice_list):
+            new_nums.extend([None] * (len(slice_list) - len(new_nums)))
+            new_offsets.extend([0] * (len(slice_list) - len(new_offsets)))
+        for i in range(len(slice_list)):
+            slice_list[i] = self._combine(slice_[i], new_nums[i], new_offsets[i])
+        for i in range(len(slice_list), len(new_nums)):
+            cur_slice = (
+                slice(new_offsets[i], new_offsets[i] + new_nums[i])
+                if new_nums[i] > 1
+                else new_offsets[i]
+            )
+            slice_list.append(cur_slice)
+        self.dataset._tensors[self.subpath][slice_list] = assign_value
 
     def _combine(self, slice_, num=None, ofs=0):
         "Combines a `slice_` with the current num and offset present in tensorview"
@@ -163,13 +167,13 @@ def _combine(self, slice_, num=None, ofs=0):
             )
             if slice_.start is None and slice_.stop is None:
                 return slice(ofs, None) if num is None else slice(ofs, ofs + num)
-            elif slice_.start is not None and slice_.stop is None:
+            elif slice_.stop is None:
                 return (
                     slice(ofs + slice_.start, None)
                     if num is None
                     else slice(ofs + slice_.start, ofs + num)
                 )
-            elif slice_.start is None and slice_.stop is not None:
+            elif slice_.start is None:
                 return slice(ofs, ofs + slice_.stop)
             else:
                 return slice(ofs + slice_.start, ofs + slice_.stop)
@@ -208,7 +212,7 @@ def slice_fill(self, slice_):
             elif offset < len(slice_):
                 new_slice_.append(slice_[offset])
                 offset += 1
-        new_slice_ = new_slice_ + slice_[offset:]
+        new_slice_ += slice_[offset:]
         return new_slice_
 
     def __repr__(self):
@@ -249,3 +253,9 @@ def chunksize(self):
     @property
     def is_dynamic(self):
         return self.dataset._tensors[self.subpath].is_dynamic
+
+    def disable_lazy(self):
+        self.lazy = False
+
+    def enable_lazy(self):
+        self.lazy = True
diff --git a/hub/api/tests/test_converters.py b/hub/api/tests/test_converters.py
index a4f1441931..9dde363ce8 100644
--- a/hub/api/tests/test_converters.py
+++ b/hub/api/tests/test_converters.py
@@ -83,6 +83,35 @@ def test_to_from_tensorflow():
         assert (res_ds["label", "d", "e", i].numpy() == i * np.ones((5, 3))).all()
 
 
+@pytest.mark.skipif(not tensorflow_loaded(), reason="requires tensorflow to be loaded")
+def test_to_from_tensorflow_datasetview():
+    my_schema = {
+        "image": Tensor((10, 1920, 1080, 3), "uint8"),
+        "label": {
+            "a": Tensor((100, 200), "int32"),
+            "b": Tensor((100, 400), "int64"),
+            "c": Tensor((5, 3), "uint8"),
+            "d": {"e": Tensor((5, 3), "uint8")},
+            "f": "float",
+        },
+    }
+
+    ds = hub.Dataset(
+        schema=my_schema, shape=(10,), url="./data/test_from_tf/ds4", mode="w"
+    )
+    for i in range(10):
+        ds["label", "d", "e", i] = i * np.ones((5, 3))
+    dsv = ds[5:]
+    tds = dsv.to_tensorflow()
+    out_ds = hub.Dataset.from_tensorflow(tds)
+    res_ds = out_ds.store(
+        "./data/test_from_tf/ds6", length=5
+    )  # generator has no length, argument needed
+
+    for i in range(5):
+        assert (res_ds["label", "d", "e", i].numpy() == (5 + i) * np.ones((5, 3))).all()
+
+
 @pytest.mark.skipif(not pytorch_loaded(), reason="requires pytorch to be loaded")
 def test_to_pytorch():
     import torch
@@ -126,7 +155,7 @@ def transform(data):
         label = recursive_torch_tensor(label)
         return (image, label)
 
-    dst = ds.to_pytorch(Transform=transform, inplace=False)
+    dst = ds.to_pytorch(transform=transform, inplace=False)
     dl = torch.utils.data.DataLoader(
         dst,
         batch_size=1,
@@ -145,6 +174,69 @@ def transform(data):
         assert type(d) == tuple
 
 
+@pytest.mark.skipif(not pytorch_loaded(), reason="requires pytorch to be loaded")
+def test_to_pytorch_datasetview():
+    import torch
+
+    my_schema = {
+        "image": Tensor((10, 1920, 1080, 3), "uint8"),
+        "label": {
+            "a": Tensor((100, 200), "int32"),
+            "b": Tensor((100, 400), "int64"),
+            "c": Tensor((5, 3), "uint8"),
+            "d": {"e": Tensor((5, 3), "uint8")},
+            "f": "float",
+        },
+    }
+    ds = hub.Dataset(
+        schema=my_schema, shape=(10,), url="./data/test_from_tf/ds5", mode="w"
+    )
+    for i in range(10):
+        ds["label", "d", "e", i] = i * np.ones((5, 3))
+    # pure conversion
+    dsv = ds[3:]
+    ptds = dsv.to_pytorch()
+    dl = torch.utils.data.DataLoader(
+        ptds,
+        batch_size=1,
+    )
+    for i, batch in enumerate(dl):
+        assert (batch["label"]["d"]["e"].numpy() == (3 + i) * np.ones((5, 3))).all()
+
+    # with transforms and inplace=False
+    def recursive_torch_tensor(label):
+        for key, value in label.items():
+            if type(value) is dict:
+                label[key] = recursive_torch_tensor(value)
+            else:
+                label[key] = torch.tensor(value)
+        return label
+
+    def transform(data):
+        image = torch.tensor(data["image"])
+        label = data["label"]
+        label = recursive_torch_tensor(label)
+        return (image, label)
+
+    dst = dsv.to_pytorch(transform=transform, inplace=False)
+    dl = torch.utils.data.DataLoader(
+        dst,
+        batch_size=1,
+    )
+    for i, batch in enumerate(dl):
+        assert (batch[1]["d"]["e"].numpy() == (3 + i) * np.ones((5, 3))).all()
+
+    # output_type = list
+    dst = ds.to_pytorch(output_type=list)
+    for i, d in enumerate(dst):
+        assert type(d) == list
+
+    # output_type = tuple
+    dst = ds.to_pytorch(output_type=tuple)
+    for i, d in enumerate(dst):
+        assert type(d) == tuple
+
+
 @pytest.mark.skipif(not pytorch_loaded(), reason="requires pytorch to be loaded")
 def test_from_pytorch():
     from torch.utils.data import Dataset
@@ -215,7 +307,6 @@ def test_to_from_pytorch():
 
 
 if __name__ == "__main__":
-
     with Timer("Test Converters"):
         with Timer("from MNIST"):
             test_from_tfds_mnist()
diff --git a/hub/api/tests/test_dataset.py b/hub/api/tests/test_dataset.py
index 3b3ffe2694..5be3612430 100644
--- a/hub/api/tests/test_dataset.py
+++ b/hub/api/tests/test_dataset.py
@@ -320,17 +320,33 @@ def test_dataset_azure():
 
 def test_datasetview_slicing():
     dt = {"first": Tensor((100, 100))}
-    ds = Dataset(schema=dt, shape=(20,), url="./data/test/model", mode="w")
-
+    ds = Dataset(
+        schema=dt, shape=(20,), url="./data/test/datasetview_slicing", mode="w"
+    )
     assert ds["first", 0].numpy().shape == (100, 100)
     assert ds["first", 0:1].numpy().shape == (1, 100, 100)
     assert ds[0]["first"].numpy().shape == (100, 100)
     assert ds[0:1]["first"].numpy().shape == (1, 100, 100)
 
 
+def test_datasetview_get_dictionary():
+    ds = Dataset(
+        schema=my_schema,
+        shape=(20,),
+        url="./data/test/datasetview_get_dictionary",
+        mode="w",
+    )
+    ds["label", 5, "a"] = 5 * np.ones((100, 200))
+    ds["label", 5, "d", "e"] = 3 * np.ones((5, 3))
+    dsv = ds[2:10]
+    dic = dsv[3, "label"]
+    assert (dic["a"].compute() == 5 * np.ones((100, 200))).all()
+    assert (dic["d"]["e"].compute() == 3 * np.ones((5, 3))).all()
+
+
 def test_tensorview_slicing():
     dt = {"first": Tensor(shape=(None, None), max_shape=(250, 300))}
-    ds = Dataset(schema=dt, shape=(20,), url="./data/test/model", mode="w")
+    ds = Dataset(schema=dt, shape=(20,), url="./data/test/tensorivew_slicing", mode="w")
     tv = ds["first", 5:6, 7:10, 9:10]
     assert tv.numpy().shape == tuple(tv.shape) == (1, 3, 1)
     tv2 = ds["first", 5:6, 7:10, 9]
@@ -395,7 +411,100 @@ def test_append_dataset():
     assert ds["second"].shape[0] == 120
 
 
+def test_dataset_compute():
+    dt = {
+        "first": Tensor(shape=(2,)),
+        "second": "float",
+        "text": Text(shape=(None,), max_shape=(12,)),
+    }
+    url = "./data/test/ds_compute"
+    ds = Dataset(schema=dt, shape=(2,), url=url, mode="w")
+    ds["text", 1] = "hello world"
+    ds["second", 0] = 3.14
+    ds["first", 0] = np.array([5, 6])
+    comp = ds.compute()
+    comp0 = comp[0]
+    assert (comp0["first"] == np.array([5, 6])).all()
+    assert comp0["second"] == 3.14
+    assert comp0["text"] == ""
+    comp1 = comp[1]
+    assert (comp1["first"] == np.array([0, 0])).all()
+    assert comp1["second"] == 0
+    assert comp1["text"] == "hello world"
+
+
+def test_dataset_view_compute():
+    dt = {
+        "first": Tensor(shape=(2,)),
+        "second": "float",
+        "text": Text(shape=(None,), max_shape=(12,)),
+    }
+    url = "./data/test/dsv_compute"
+    ds = Dataset(schema=dt, shape=(4,), url=url, mode="w")
+    ds["text", 3] = "hello world"
+    ds["second", 2] = 3.14
+    ds["first", 2] = np.array([5, 6])
+    dsv = ds[2:]
+    comp = dsv.compute()
+    comp0 = comp[0]
+    assert (comp0["first"] == np.array([5, 6])).all()
+    assert comp0["second"] == 3.14
+    assert comp0["text"] == ""
+    comp1 = comp[1]
+    assert (comp1["first"] == np.array([0, 0])).all()
+    assert comp1["second"] == 0
+    assert comp1["text"] == "hello world"
+
+
+def test_dataset_lazy():
+    dt = {
+        "first": Tensor(shape=(2,)),
+        "second": "float",
+        "text": Text(shape=(None,), max_shape=(12,)),
+    }
+    url = "./data/test/ds_lazy"
+    ds = Dataset(schema=dt, shape=(2,), url=url, mode="w", lazy=False)
+    ds["text", 1] = "hello world"
+    ds["second", 0] = 3.14
+    ds["first", 0] = np.array([5, 6])
+    assert ds["text", 1] == "hello world"
+    assert ds["second", 0] == 3.14
+    assert (ds["first", 0] == np.array([5, 6])).all()
+
+
+def test_dataset_view_lazy():
+    dt = {
+        "first": Tensor(shape=(2,)),
+        "second": "float",
+        "text": Text(shape=(None,), max_shape=(12,)),
+    }
+    url = "./data/test/dsv_lazy"
+    ds = Dataset(schema=dt, shape=(4,), url=url, mode="w", lazy=False)
+    ds["text", 3] = "hello world"
+    ds["second", 2] = 3.14
+    ds["first", 2] = np.array([5, 6])
+    dsv = ds[2:]
+    assert dsv["text", 1] == "hello world"
+    assert dsv["second", 0] == 3.14
+    assert (dsv["first", 0] == np.array([5, 6])).all()
+
+
+def test_datasetview_repr():
+    dt = {
+        "first": Tensor(shape=(2,)),
+        "second": "float",
+        "text": Text(shape=(None,), max_shape=(12,)),
+    }
+    url = "./data/test/dsv_repr"
+    ds = Dataset(schema=dt, shape=(9,), url=url, mode="w", lazy=False)
+    dsv = ds[2:]
+    print_text = "DatasetView(Dataset(schema=SchemaDict({'first': Tensor(shape=(2,), dtype='float64'), 'second': 'float64', 'text': Text(shape=(None,), dtype='int64', max_shape=(12,))})url='./data/test/dsv_repr', shape=(9,), mode='w'), slice=slice(2, 9, None))"
+    assert dsv.__repr__() == print_text
+
+
 if __name__ == "__main__":
+    test_datasetview_repr()
+    test_datasetview_get_dictionary()
     test_tensorview_slicing()
     test_datasetview_slicing()
     test_dataset()
@@ -404,4 +513,8 @@ def test_append_dataset():
     test_dataset2()
     test_text_dataset()
     test_text_dataset_tokenizer()
+    test_dataset_compute()
+    test_dataset_view_compute()
+    test_dataset_lazy()
+    test_dataset_view_lazy()
     test_dataset_hub()
diff --git a/hub/compute/ray.py b/hub/compute/ray.py
index 1951415d7b..bedddbdfa6 100644
--- a/hub/compute/ray.py
+++ b/hub/compute/ray.py
@@ -3,13 +3,13 @@
 from hub.api.datasetview import DatasetView
 from hub.utils import batchify
 from hub.compute import Transform
-from typing import Iterable
+from typing import Iterable, Iterator
 from hub.exceptions import ModuleNotInstalledException
 from hub.api.sharded_datasetview import ShardedDatasetView
 import hub
 
 
-def remote(template, **kwargs):
+def empty_remote(template, **kwargs):
     """
     remote template
     """
@@ -28,7 +28,7 @@ def inner(**kwargs):
 
     remote = ray.remote
 except Exception:
-    pass
+    remote = empty_remote
 
 
 class RayTransform(Transform):
@@ -36,24 +36,29 @@ def __init__(self, func, schema, ds, scheduler="ray", workers=1, **kwargs):
         super(RayTransform, self).__init__(
             func, schema, ds, scheduler="single", workers=workers, **kwargs
         )
+        self.workers = workers
         if "ray" not in sys.modules:
             raise ModuleNotInstalledException("ray")
 
         if not ray.is_initialized():
-            ray.init()
+            ray.init(local_mode=True)
 
     @remote
-    def _func_argd(_func, index, _ds, schema, **kwargs):
+    def _func_argd(_func, index, _ds, schema, kwargs):
         """
         Remote wrapper for user defined function
         """
+
         if isinstance(_ds, Dataset) or isinstance(_ds, DatasetView):
             _ds.squeeze_dim = False
 
         item = _ds[index]
-        item = _func(item, **kwargs)
-        # item = Transform._flatten(item, schema)
+        if isinstance(item, DatasetView) or isinstance(item, Dataset):
+            item = item.compute()
+
+        item = _func(0, item)
         item = Transform._flatten_dict(item, schema=schema)
+
         return list(item.values())
 
     def store(
@@ -84,7 +89,6 @@ def store(
         ds: hub.Dataset
             uploaded dataset
         """
-
         _ds = ds or self._ds
         if isinstance(_ds, Transform):
             _ds = _ds.store(
@@ -96,12 +100,16 @@ def store(
         num_returns = len(self._flatten_dict(self.schema, schema=self.schema).keys())
         results = [
             self._func_argd.options(num_returns=num_returns).remote(
-                self._func, el, _ds, schema=self.schema, **self.kwargs
+                self.call_func, el, _ds, schema=self.schema, kwargs=self.kwargs
             )
             for el in range(len(_ds))
         ]
 
+        if num_returns == 1:
+            results = [[r] for r in results]
+
         results = self._split_list_to_dicts(results)
+
         ds = self.upload(results, url=url, token=token, progressbar=progressbar)
         return ds
 
@@ -109,18 +117,40 @@ def store(
     def upload_chunk(i_batch, key, ds):
         """
         Remote function to upload a chunk
+        Returns the shape of dynamic tensor to upload all in once after upload is completed
+
+        Parameters
+        ----------
+        i_batch: Tuple
+            Tuple composed of (index, batch)
+        key: str
+            Key of the tensor
+        ds:
+            Dataset to set to upload
+        Returns
+        ----------
+        (key, slice_, shape) to set the shape later
+
         """
         i, batch = i_batch
-        length = len(batch)
-
         if not isinstance(batch, dict) and isinstance(batch[0], ray.ObjectRef):
             batch = ray.get(batch)
+            # FIXME an ugly hack to unwrap elements with a schema that has one tensor
+            num_returns = len(
+                Transform._flatten_dict(ds.schema.dict_, schema=ds.schema.dict_).keys()
+            )
+            if num_returns == 1:
+                batch = [item for sublist in batch for item in sublist]
 
-        # TODO some sort of syncronizer across nodes
-        if length != 1:
-            ds[key, i * length : (i + 1) * length] = batch
-        else:
-            ds[key, i * length] = batch[0]
+        shape = None
+        length = len(batch)
+
+        slice_ = slice(i * length, (i + 1) * length)
+        if ds[key].is_dynamic:
+            shape = ds._tensors[f"/{key}"].get_shape_from_value([slice_], batch)
+        ds[key, slice_] = batch
+
+        return (key, [slice_], shape)
 
     def upload(self, results, url: str, token: dict, progressbar: bool = True):
         """Batchified upload of results.
@@ -139,11 +169,15 @@ def upload(self, results, url: str, token: dict, progressbar: bool = True):
         ds: hub.Dataset
             Uploaded dataset
         """
-        shape = (len(list(results.values())[0]),)
+        if len(list(results.values())) == 0:
+            shape = (0,)
+        else:
+            shape = (len(list(results.values())[0]),)
+
         ds = Dataset(
             url,
             mode="w",
-            shape=shape,  # unkownn
+            shape=shape,
             schema=self.schema,
             token=token,
             cache=False,
@@ -151,24 +185,57 @@ def upload(self, results, url: str, token: dict, progressbar: bool = True):
 
         tasks = []
         for key, value in results.items():
+
             length = ds[key].chunksize[0]
             batched_values = batchify(value, length)
-
             chunk_id = list(range(len(batched_values)))
             index_batched_values = list(zip(chunk_id, batched_values))
+
+            ds._tensors[f"/{key}"].disable_dynamicness()
+
             results = [
                 self.upload_chunk.remote(el, key=key, ds=ds)
                 for el in index_batched_values
             ]
             tasks.extend(results)
 
-        ray.get(tasks)
+        results = ray.get(tasks)
+        self.set_dynamic_shapes(results, ds)
         ds.commit()
         return ds
 
+    def set_dynamic_shapes(self, results, ds):
+        """
+        Sets shapes for dynamic tensors after the dataset is uploaded
+
+        Parameters
+        ----------
+        results: Tuple
+            results from uploading each chunk which includes (key, slice, shape) tuple
+        ds:
+            Dataset to set the shapes to
+        Returns
+        ----------
+        """
+
+        shapes = {}
+        for (key, slice_, value) in results:
+            if not ds[key].is_dynamic:
+                continue
+
+            if key not in shapes:
+                shapes[key] = []
+            shapes[key].append((slice_, value))
+
+        for key, value in shapes.items():
+            ds._tensors[f"/{key}"].enable_dynamicness()
+
+            for (slice_, shape) in shapes[key]:
+                ds._tensors[f"/{key}"].set_dynamic_shape(slice_, shape)
+
 
 class TransformShard:
-    def __init__(self, ds, func, schema, **kwargs):
+    def __init__(self, ds, func, schema, kwargs):
 
         if isinstance(ds, Dataset) or isinstance(ds, DatasetView):
             ds.squeeze_dim = False
@@ -185,9 +252,14 @@ def __call__(self, ids):
         """
         for index in ids:
             item = self._ds[index]
-            item = self._func(item, **self.kwargs)
+            if isinstance(item, DatasetView) or isinstance(item, Dataset):
+                item = item.compute()
+
+            items = self._func(0, item)
+            if not isinstance(items, list):
+                items = [items]
 
-            for item in Transform._unwrap(item):
+            for item in items:
                 yield Transform._flatten_dict(item, schema=self.schema)
 
 
@@ -228,8 +300,10 @@ def store(
                 progressbar=progressbar,
             )
 
-        results = ray.util.iter.from_range(len(_ds), num_shards=4).transform(
-            TransformShard(ds=_ds, func=self._func, schema=self.schema, **self.kwargs)
+        results = ray.util.iter.from_range(len(_ds), num_shards=self.workers).transform(
+            TransformShard(
+                ds=_ds, func=self.call_func, schema=self.schema, kwargs=self.kwargs
+            )
         )
 
         @remote
@@ -239,7 +313,7 @@ def upload_shard(i, shard_results):
             """
             shard_results = self._split_list_to_dicts(shard_results)
 
-            if len(shard_results) == 0:
+            if len(shard_results) == 0 or len(list(shard_results.values())[0]) == 0:
                 return None
 
             ds = self.upload(
@@ -270,19 +344,9 @@ def merge_sharded_dataset(
         """
         sharded_ds = ShardedDatasetView(datasets)
 
-        def identity(sample):
-            d = {}
-            for k in sample.keys:
-                v = sample[k]
-                if not isinstance(v, dict):
-                    d[k] = v.compute()
-                else:
-                    d[k] = identity(v)
-            return d
-
         @hub.transform(schema=self.schema, scheduler="ray")
         def transform_identity(sample):
-            return identity(sample)
+            return sample
 
         ds = transform_identity(sharded_ds).store(
             url,
diff --git a/hub/compute/tests/test_ray.py b/hub/compute/tests/test_ray.py
index 767fc0df12..bcfba342f9 100644
--- a/hub/compute/tests/test_ray.py
+++ b/hub/compute/tests/test_ray.py
@@ -2,6 +2,7 @@
 from hub.utils import ray_loaded
 from hub.schema import Tensor, Text
 import pytest
+from hub.compute.ray import empty_remote
 
 import numpy as np
 
@@ -18,6 +19,79 @@
 }
 
 
+def test_wrapper():
+    @empty_remote
+    def a(x):
+        return x
+
+    assert a(5)() == 5
+
+
+@pytest.mark.skipif(
+    not ray_loaded(),
+    reason="requires ray to be loaded",
+)
+def test_ray_simple():
+    schema = {"var": "float"}
+
+    @hub.transform(schema=schema, scheduler="ray")
+    def process(item):
+        return {"var": 1}
+
+    ds = process([1, 2, 3]).store("./data/somedataset")
+    assert ds["var", 0].compute() == 1
+
+
+@pytest.mark.skipif(
+    not ray_loaded(),
+    reason="requires ray to be loaded",
+)
+def test_ray_non_dynamic():
+    schema = {
+        "var": Tensor(shape=(2, 2), dtype="uint8"),
+        "var2": Tensor(shape=(2, 2), dtype="uint8"),
+    }
+
+    @hub.transform(schema=schema, scheduler="ray_generator")
+    def process(item):
+        return [{"var": np.ones((2, 2)), "var2": np.ones((2, 2))} for i in range(2)]
+
+    ds = process([1, 2, 3]).store("./data/somedataset")
+    assert ds["var", 0].compute().shape[0] == 2
+
+
+@pytest.mark.skipif(
+    not ray_loaded(),
+    reason="requires ray to be loaded",
+)
+def test_ray_dynamic():
+    schema = {"var": Tensor(shape=(None, None), max_shape=(2, 2), dtype="uint8")}
+
+    @hub.transform(schema=schema, scheduler="ray_generator")
+    def process(item):
+        return {"var": np.ones((1, 2))}
+
+    ds = process([1, 2, 3]).store("./data/somedataset")
+    assert ds["var", 0].compute().shape[0] == 1
+
+
+@pytest.mark.skipif(
+    not ray_loaded(),
+    reason="requires ray to be loaded",
+)
+def test_ray_simple_generator():
+    schema = {"var": "float"}
+
+    @hub.transform(schema=schema, scheduler="ray_generator")
+    def process(item):
+        items = [{"var": item} for i in range(item)]
+        return items
+
+    ds = process([0, 1, 2, 3]).store("./data/somegeneratordataset")
+    assert ds["var", 0].compute() == 1
+    assert ds.shape[0] == 6
+
+
 @pytest.mark.skipif(
     not ray_loaded(),
     reason="requires ray to be loaded",
@@ -39,10 +113,10 @@ def test_pipeline_ray():
     @hub.transform(schema=my_schema, scheduler="ray")
     def my_transform(sample, multiplier: int = 2):
         return {
-            "image": sample["image"].compute() * multiplier,
-            "label": sample["label"].compute(),
+            "image": sample["image"] * multiplier,
+            "label": sample["label"],
             "confidence": {
-                "confidence": sample["confidence/confidence"].compute() * multiplier
+                "confidence": sample["confidence"]["confidence"] * multiplier
             },
         }
 
@@ -72,8 +146,8 @@ def test_ray_pipeline_multiple():
     def dynamic_transform(sample, multiplier: int = 2):
         return [
             {
-                "image": sample["image"].compute() * multiplier,
-                "label": sample["label"].compute(),
+                "image": sample["image"] * multiplier,
+                "label": sample["label"],
             }
             for _ in range(4)
         ]
@@ -88,4 +162,9 @@ def dynamic_transform(sample, multiplier: int = 2):
 
 
 if __name__ == "__main__":
-    test_ray_pipeline_multiple()
+    # test_ray_simple()
+    # test_ray_non_dynamic()
+    test_ray_dynamic()
+    # test_ray_simple_generator()
+    # test_pipeline_ray()
+    # test_ray_pipeline_multiple()
diff --git a/hub/compute/tests/test_transform.py b/hub/compute/tests/test_transform.py
index 8610cdf116..0712eae2c9 100644
--- a/hub/compute/tests/test_transform.py
+++ b/hub/compute/tests/test_transform.py
@@ -30,9 +30,9 @@ def test_pipeline_basic():
     @hub.transform(schema=my_schema)
     def my_transform(sample, multiplier: int = 2):
         return {
-            "image": sample["image"].compute() * multiplier,
-            "label": sample["label"].compute(),
-            "confidence": sample["confidence"].compute() * multiplier,
+            "image": sample["image"] * multiplier,
+            "label": sample["label"],
+            "confidence": sample["confidence"] * multiplier,
         }
 
     out_ds = my_transform(ds, multiplier=2)
@@ -77,7 +77,7 @@ def test_threaded():
 
     @hub.transform(schema=schema, scheduler="threaded", workers=2)
     def create_classification_dataset(sample):
-        ts = sample["image"].numpy()
+        ts = sample["image"]
         return [
             {
                 "image": ts,
@@ -109,8 +109,8 @@ def test_pipeline_dynamic():
     @hub.transform(schema=dynamic_schema)
     def dynamic_transform(sample, multiplier: int = 2):
         return {
-            "image": sample["image"].compute() * multiplier,
-            "label": sample["label"].compute(),
+            "image": sample["image"] * multiplier,
+            "label": sample["label"],
         }
 
     out_ds = dynamic_transform(ds, multiplier=4).store(
@@ -137,8 +137,8 @@ def test_pipeline_multiple():
     def dynamic_transform(sample, multiplier: int = 2):
         return [
             {
-                "image": sample["image"].compute() * multiplier,
-                "label": sample["label"].compute(),
+                "image": sample["image"] * multiplier,
+                "label": sample["label"],
             }
             for i in range(4)
         ]
@@ -206,9 +206,9 @@ def test_pipeline():
         @hub.transform(schema=my_schema)
         def my_transform(sample, multiplier: int = 2):
             return {
-                "image": sample["image"].compute() * multiplier,
-                "label": sample["label"].compute(),
-                "confidence": sample["confidence"].compute() * multiplier,
+                "image": sample["image"] * multiplier,
+                "label": sample["label"],
+                "confidence": sample["confidence"] * multiplier,
             }
 
         out_ds = my_transform(ds, multiplier=2)
@@ -218,6 +218,59 @@ def my_transform(sample, multiplier: int = 2):
         assert (out_ds["image", 0].compute() == 4).all()
 
 
+def test_stacked_transform():
+    schema = {"test": Tensor((2, 2), dtype="uint8")}
+
+    @hub.transform(schema=schema)
+    def multiply_transform(sample, multiplier=1, times=1):
+        if times == 1:
+            return {"test": multiplier * sample["test"]}
+        else:
+            return [{"test": multiplier * sample["test"]} for i in range(times)]
+
+    ds = hub.Dataset("./data/stacked_transform", mode="w", shape=(5,), schema=schema)
+    for i in range(5):
+        ds["test", i] = np.ones((2, 2))
+    ds1 = multiply_transform(ds, multiplier=2, times=5)
+    ds2 = multiply_transform(ds1, multiplier=3, times=2)
+    ds3 = multiply_transform(ds2, multiplier=5, times=3)
+    ds4 = ds3.store("./data/stacked_transform_2")
+    assert len(ds4) == 150
+    assert (ds4["test", 0].compute() == 30 * np.ones((2, 2))).all()
+
+
+def test_zero_sample_transform():
+    schema = {"test": Tensor((None, None), dtype="uint8", max_shape=(10, 10))}
+
+    @hub.transform(schema=schema)
+    def my_transform(sample):
+        if sample % 5 == 0:
+            return []
+        else:
+            return {"test": (sample % 5) * np.ones((5, 5))}
+
+    ds = my_transform([i for i in range(30)])
+    ds2 = ds.store("./data/transform_zero_sample", sample_per_shard=5)
+    assert len(ds2) == 24
+    for i, item in enumerate(ds2):
+        assert (item["test"].compute() == (((i % 4) + 1) * np.ones((5, 5)))).all()
+
+
+def test_mutli_sample_transform():
+    schema = {"test": Tensor((None, None), dtype="uint8", max_shape=(10, 10))}
+
+    @hub.transform(schema=schema)
+    def my_transform(sample):
+        d = {"test": sample * np.ones((5, 5))}
+        return [d, d]
+
+    ds = my_transform([i for i in range(32)])
+    ds2 = ds.store("./data/transform_zero_sample", sample_per_shard=5)
+    assert len(ds2) == 64
+    for i, item in enumerate(ds2):
+        assert (item["test"].compute() == (i // 2) * np.ones((5, 5))).all()
+
+
 def benchmark(sample_size=100, width=1000, channels=4, dtype="int8"):
     numpy_arr = np.zeros((sample_size, width, width, channels), dtype=dtype)
     zarr_fs = zarr.zeros(
diff --git a/hub/compute/transform.py b/hub/compute/transform.py
index 0d1c690d4f..efad597f51 100644
--- a/hub/compute/transform.py
+++ b/hub/compute/transform.py
@@ -1,7 +1,6 @@
 import zarr
 import numpy as np
 import math
-from psutil import virtual_memory
 from typing import Dict, Iterable
 from hub.api.dataset import Dataset
 from tqdm import tqdm
@@ -38,6 +37,7 @@ def prod(shp):
             return res
 
         samples = min(samples, (16 * 1024 * 1024 * 8) // (prod(shp) * sz))
+    samples = max(samples, 1)
     return samples * workers
 
 
@@ -68,6 +68,17 @@ def __init__(
         self.kwargs = kwargs
         self.workers = workers
 
+        if isinstance(self._ds, Transform):
+            self.base_ds = self._ds.base_ds
+            self._func = self._ds._func[:]
+            self._func.append(func)
+            self.kwargs = self._ds.kwargs[:]
+            self.kwargs.append(kwargs)
+        else:
+            self.base_ds = ds
+            self._func = [func]
+            self.kwargs = [kwargs]
+
         if scheduler == "threaded" or (scheduler == "single" and workers > 1):
             self.map = ThreadPool(nodes=workers).map
         elif scheduler == "processed":
@@ -85,6 +96,48 @@ def __init__(
                 f"Scheduler {scheduler} not understood, please use 'single', 'threaded', 'processed'"
             )
 
+    def __len__(self):
+        return self.shape[0]
+
+    def __getitem__(self, slice_):
+        """| Get an item to be computed without iterating on the whole dataset.
+        | Creates a dataset view, then a temporary dataset to apply the transform.
+        Parameters:
+        ----------
+        slice_: slice
+            Gets a slice or slices from dataset
+        """
+        if not isinstance(slice_, abc.Iterable) or isinstance(slice_, str):
+            slice_ = [slice_]
+
+        slice_ = list(slice_)
+        subpath, slice_list = slice_split(slice_)
+
+        if len(slice_list) == 0:
+            slice_list = [slice(None, None, None)]
+
+        num, ofs = slice_extract_info(slice_list[0], self.shape[0])
+
+        ds_view = DatasetView(
+            dataset=self._ds,
+            num_samples=num,
+            offset=ofs,
+            squeeze_dim=isinstance(slice_list[0], int),
+        )
+
+        path = posixpath.expanduser("~/.activeloop/tmparray")
+        new_ds = self.store(path, length=num, ds=ds_view, progressbar=False)
+
+        index = 1 if len(slice_) > 1 else 0
+        slice_[index] = (
+            slice(None, None, None) if not isinstance(slice_list[0], int) else 0
+        )  # Get all shape dimension since we already sliced
+        return new_ds[slice_]
+
+    def __iter__(self):
+        for index in range(len(self)):
+            yield self[index]
+
     @classmethod
     def _flatten_dict(self, d: Dict, parent_key="", schema=None):
         """| Helper function to flatten dictionary of a recursive tensor
@@ -136,6 +189,19 @@ def dtype_from_path(cls, path, schema):
             cur_type = cur_type.dict_
         return cur_type[path[-1]]
 
+    @classmethod
+    def _unwrap(cls, results):
+        """
+        If there is any list then unwrap it into its elements
+        """
+        items = []
+        for r in results:
+            if isinstance(r, dict):
+                items.append(r)
+            else:
+                items.extend(r)
+        return items
+
     def _split_list_to_dicts(self, xs):
         """| Helper function that transform list of dicts into dicts of lists
 
@@ -160,6 +226,17 @@ def _split_list_to_dicts(self, xs):
                     xs_new[key] = [value]
         return xs_new
 
+    def _pbar(self, show: bool = True):
+        """
+        Returns a progress bar, if empty then it function does nothing
+        """
+
+        def _empty_pbar(xs, **kwargs):
+            return xs
+
+        single_threaded = self.map == map
+        return tqdm if show and single_threaded else _empty_pbar
+
     def create_dataset(
         self, url: str, length: int = None, token: dict = None, public: bool = True
     ):
@@ -207,11 +284,9 @@ def upload(self, results, ds: Dataset, token: dict, progressbar: bool = True):
 
             def upload_chunk(i_batch):
                 i, batch = i_batch
-                batch_length = len(batch)
-                if batch_length != 1:
-                    ds[key, i * length : i * length + batch_length] = batch
-                else:
-                    ds[key, i * length] = batch[0]
+                length = len(batch)
+                slice_ = slice(i * length, (i + 1) * length)
+                ds[key, slice_] = batch
 
             index_batched_values = list(
                 zip(list(range(len(batched_values))), batched_values)
@@ -231,29 +306,32 @@ def upload_chunk(i_batch):
         ds.commit()
         return ds
 
-    def _pbar(self, show: bool = True):
-        """
-        Returns a progress bar, if empty then it function does nothing
-        """
+    def call_func(self, fn_index, item, as_list=False):
+        """Calls all the functions one after the other
 
-        def _empty_pbar(xs, **kwargs):
-            return xs
-
-        single_threaded = self.map == map
-        return tqdm if show and single_threaded else _empty_pbar
+        Parameters
+        ----------
+        fn_index: int
+            The index starting from which the functions need to be called
+        item:
+            The item on which functions need to be applied
+        as_list: bool, optional
+            If true then treats the item as a list.
 
-    @classmethod
-    def _unwrap(cls, results):
-        """
-        If there is any list then unwrap it into its elements
+        Returns
+        ----------
+        result:
+            The final output obtained after all transforms
         """
-        items = []
-        for r in results:
-            if isinstance(r, dict):
-                items.append(r)
+        result = item
+        if fn_index < len(self._func):
+            if as_list:
+                result = [self.call_func(fn_index, it) for it in result]
             else:
-                items.extend(r)
-        return items
+                result = self._func[fn_index](result, **self.kwargs[fn_index])
+                result = self.call_func(fn_index + 1, result, isinstance(result, list))
+        result = self._unwrap(result) if isinstance(result, list) else result
+        return result
 
     def store_shard(self, ds_in: Iterable, ds_out: Dataset, offset: int, token=None):
         """
@@ -261,7 +339,12 @@ def store_shard(self, ds_in: Iterable, ds_out: Dataset, offset: int, token=None)
         """
 
         def _func_argd(item):
-            return self._func(item, **self.kwargs)
+            if isinstance(item, DatasetView) or isinstance(item, Dataset):
+                item = item.numpy()
+            result = self.call_func(
+                0, item
+            )  # If the iterable obtained from iterating ds_in is a list, it is not treated as list
+            return result
 
         ds_in = list(ds_in)
         results = self.map(
@@ -300,7 +383,7 @@ def store(
         length: int = None,
         ds: Iterable = None,
         progressbar: bool = True,
-        sample_per_shard: bool = None,
+        sample_per_shard: int = None,
         public: bool = True,
     ):
         """| The function to apply the transformation for each element in batchified manner
@@ -329,13 +412,7 @@ def store(
             uploaded dataset
         """
 
-        ds_in = ds or self._ds
-        if isinstance(ds_in, Transform):
-            ds_in = ds_in.store(
-                "{}_{}".format(url, ds_in._func.__name__),
-                token=token,
-                progressbar=progressbar,
-            )
+        ds_in = ds or self.base_ds
 
         # compute shard length
         if sample_per_shard is None:
@@ -373,57 +450,13 @@ def batchify_generator(iterator: Iterable, size: int):
             for ds_in_shard in batchify_generator(ds_in, n_samples):
                 n_results = self.store_shard(ds_in_shard, ds_out, start, token=token)
                 total += n_results
-                pbar.update(n_results)
-                if n_results < n_samples or n_results == 0:
-                    break
-                start += n_samples
+                pbar.update(len(ds_in_shard))
+                start += n_results
 
         ds_out.resize_shape(total)
         ds_out.commit()
         return ds_out
 
-    def __len__(self):
-        return self.shape[0]
-
-    def __getitem__(self, slice_):
-        """| Get an item to be computed without iterating on the whole dataset.
-        | Creates a dataset view, then a temporary dataset to apply the transform.
-        Parameters:
-        ----------
-        slice_: slice
-            Gets a slice or slices from dataset
-        """
-        if not isinstance(slice_, abc.Iterable) or isinstance(slice_, str):
-            slice_ = [slice_]
-
-        slice_ = list(slice_)
-        subpath, slice_list = slice_split(slice_)
-
-        if len(slice_list) == 0:
-            slice_list = [slice(None, None, None)]
-
-        num, ofs = slice_extract_info(slice_list[0], self.shape[0])
-
-        ds_view = DatasetView(
-            dataset=self._ds,
-            num_samples=num,
-            offset=ofs,
-            squeeze_dim=isinstance(slice_list[0], int),
-        )
-
-        path = posixpath.expanduser("~/.activeloop/tmparray")
-        new_ds = self.store(path, length=num, ds=ds_view, progressbar=False)
-
-        index = 1 if len(slice_) > 1 else 0
-        slice_[index] = (
-            slice(None, None, None) if not isinstance(slice_list[0], int) else 0
-        )  # Get all shape dimension since we already sliced
-        return new_ds[slice_]
-
-    def __iter__(self):
-        for index in range(len(self)):
-            yield self[index]
-
     @property
     def shape(self):
         return self._ds.shape
diff --git a/hub/store/dynamic_tensor.py b/hub/store/dynamic_tensor.py
index 731fc6661c..11af6bfbf4 100644
--- a/hub/store/dynamic_tensor.py
+++ b/hub/store/dynamic_tensor.py
@@ -343,14 +343,28 @@ def get_shape(self, slice_):
             return final_shapes  # returns 2D np array
 
     def set_shape(self, slice_, value):
-        """Sets the shape of the slice of tensor"""
+        """
+        Set shape of the dynamic tensor given value
+        """
         if not self._enabled_dynamicness:
             return
+
+        new_shape = self.get_shape_from_value(slice_, value)
+        self.set_dynamic_shape(slice_, new_shape)
+
+    def set_dynamic_shape(self, slice_, shape):
+        """
+        Set shape from the shape directly
+        """
+        self._dynamic_tensor[slice_[0]] = shape
+
+    def get_shape_from_value(self, slice_, value):
+        """
+        create shape for multiple elements
+        """
         if isinstance(slice_[0], int):
-            new_shape = self.create_shape(slice_, value)
-            self._dynamic_tensor[slice_[0]] = new_shape = np.maximum(
-                self._dynamic_tensor[slice_[0]], new_shape
-            )
+            new_shapes = self.create_shape(slice_, value)
+            new_shapes = np.maximum(self._dynamic_tensor[slice_[0]], new_shapes)
         else:
             start = slice_[0].start if slice_[0].start is not None else 0
             stop = (
@@ -362,7 +376,7 @@ def set_shape(self, slice_, value):
                 new_shape = self.create_shape([i] + slice_[1:], value[i - start])
                 new_shape = np.maximum(dt[i - start], new_shape)
                 new_shapes.append(new_shape)
-            self._dynamic_tensor[slice_[0]] = new_shapes
+        return new_shapes
 
     def create_shape(self, slice_, value):
         assert isinstance(slice_[0], int)


Code review comments:
diff_hunk: @@ -84,7 +89,6 @@ def store(
         ds: hub.Dataset
             uploaded dataset
         """
-
         _ds = ds or self._ds
path: hub/compute/ray.py
text: @author:
I think it should be _ds = ds or self.base_ds
Then we can skip the next line that stores the intermediate transform as done in transform.

@user1:
Do you mean to skip if conditioning as a whole after replacing `_ds = ds or self._ds`? 

@author:
Yes. We won't need to store the intermediate transform any more. Take a look at transform.py