FEAT-modin-project#5809: New implementation of the Ray lazy execution…

… queue

Signed-off-by: Andrey Pavlenko <andrey.a.pavlenko@gmail.com>

modin/core/execution/ray/common/utils.py

@@ -322,6 +322,9 @@ def deconstruct_call_queue(call_queue):
     kw_value_lengths = []
     unfolded_queue = []
     for call in call_queue:
+        if isinstance(call, int):
+            unfolded_queue.append(call)
+            continue
         unfolded_queue.append(call[0])
         unfolded_queue.extend(call[1])
         arg_lengths.append(len(call[1]))
@@ -379,8 +382,15 @@ def take_n_items(n):
         return res
 
     call_queue = []
-    for i in range(num_funcs):
+    idx_shift = 0
+    for idx in range(num_funcs):
         func = take_n_items(1)[0]
+        if isinstance(func, int):
+            call_queue.append(func)
+            idx_shift += 1
+            continue
+
+        i = idx - idx_shift
         args = take_n_items(arg_lengths[i])
         kw_keys = take_n_items(kw_key_lengths[i])
         kwargs = {}