Optimize dynamic subqueries using the same task index

Author: snower

syncanysql/calculaters/query_tasker_calculater.py

@@ -26,15 +26,15 @@ def start(self, tasker, loader, arguments, task_config, **kwargs):
                 if key[:knl] != kn:
                     continue
                 self.arguments[key[knl:]] = value
-            self.create_executor_tasker()
+            self.create_executor_tasker(current_tasker)
         finally:
             _thread_local.current_tasker = current_tasker
 
     def calculate(self, primary_keys, query, task_config, *args):
         current_tasker = _thread_local.current_tasker
         try:
             if self.executor is None:
-                self.create_executor_tasker()
+                self.create_executor_tasker(current_tasker)
             with self.executor as executor:
                 for exp, values in query["filters"].items():
                     if not exp:
@@ -54,7 +54,7 @@ def calculate(self, primary_keys, query, task_config, *args):
             self.executor, self.tasker = None, None
             _thread_local.current_tasker = current_tasker
 
-    def create_executor_tasker(self):
+    def create_executor_tasker(self, current_tasker):
         from ..executor import Executor
         from ..taskers.query import QueryTasker
 
@@ -64,6 +64,9 @@ def create_executor_tasker(self):
             config["output"] = "&.--.__queryTasker_" + str(id(executor)) + "::" + config["output"].split("::")[-1]
             config["name"] = config["name"] + "#queryTasker"
             tasker = QueryTasker(config, is_inner_subquery=True)
+            if current_tasker and hasattr(current_tasker, "tasker_index"):
+                setattr(tasker, "tasker_index", executor.distribute_tasker_index())
+                tasker.name = "[%s]%s" % (current_tasker.tasker_index, tasker.name)
             executor.runners.extend(tasker.start(config.get("name"), executor, executor.session_config,
                                                  executor.manager, self.arguments))
             self.tasker, self.executor = tasker, executor

syncanysql/taskers/query.py

@@ -143,8 +143,9 @@ def start(self, name, executor, session_config, manager, arguments):
             tasker.add_hooker(ReduceHooker(executor, session_config, manager, arguments,
                                            self, copy.deepcopy(self.config), batch, aggregate))
         arguments = self.compile_tasker(arguments, tasker)
-        setattr(tasker, "tasker_index", executor.distribute_tasker_index())
-        tasker.name = "[%s]%s" % (tasker.tasker_index, tasker.name)
+        if not hasattr(tasker, "tasker_index"):
+            setattr(tasker, "tasker_index", executor.distribute_tasker_index())
+            tasker.name = "[%s]%s" % (tasker.tasker_index, tasker.name)
         self.tasker, self.dependency_taskers, self.arguments = tasker, dependency_taskers, arguments
         return [self]