allow clients to prepare for job execution on thread and move query t…

…ag execution there.

dlt/common/destination/reference.py

@@ -352,6 +352,7 @@ def run_managed(
         # filepath is now moved to running
         try:
             self._state = "running"
+            self._job_client.prepare_load_job_execution(self)
             self.run()
             self._state = "completed"
         except (DestinationTerminalException, TerminalValueError) as e:
@@ -456,6 +457,10 @@ def create_load_job(
         """Creates a load job for a particular `table` with content in `file_path`"""
         pass
 
+    def prepare_load_job_execution(self, job: RunnableLoadJob) -> None:
+        """Prepare the connected job client for the execution of a load job (used for query tags in sql clients)"""
+        pass
+
     def should_truncate_table_before_load(self, table: TTableSchema) -> bool:
         return table["write_disposition"] == "replace"
 

dlt/destinations/job_client_impl.py

@@ -256,7 +256,6 @@ def create_load_job(
         self, table: TTableSchema, file_path: str, load_id: str, restore: bool = False
     ) -> LoadJob:
         """Starts SqlLoadJob for files ending with .sql or returns None to let derived classes to handle their specific jobs"""
-        self._set_query_tags_for_job(load_id, table)
         if SqlLoadJob.is_sql_job(file_path):
             # create sql load job
             return SqlLoadJob(file_path)
@@ -655,6 +654,9 @@ def _verify_schema(self) -> None:
                 logger.error(str(exception))
             raise exceptions[0]
 
+    def prepare_load_job_execution(self, job: RunnableLoadJob) -> None:
+        self._set_query_tags_for_job(load_id=job._load_id, table=job._load_table)
+
     def _set_query_tags_for_job(self, load_id: str, table: TTableSchema) -> None:
         """Sets query tags in sql_client for a job in package `load_id`, starting for a particular `table`"""
         from dlt.common.pipeline import current_pipeline