<p dir="auto">All my use cases would work better if each operator would execute everything in one transaction. Two examples:</p>
<ul dir="auto">
<li>I want to <code class="notranslate">GenericTransfer</code> a set of rows from one DB to another, and I have to create the table first in the destination DB. I feel like it'd be a lot more clean if I didn't have empty tables lying around if the insertion fails for some reason later on.</li>
<li>I want to <code class="notranslate">GenericTransfer</code> all rows from an entire table periodically to sync it from one DB to another. To do this correctly I want to clear the destination table first to make sure I end up with no duplicate rows, so I'd have a <code class="notranslate">DELETE * FROM dst_table</code> preoperator. If the insertions fail afterwards, I'd end up with no data (it would be better in most cases to fall back to the old data), and even if everything is working correctly, I'll have an empty table while the insertions as still executing.</li>
</ul>
<p dir="auto">To fix this, the relevant <code class="notranslate">DbApiHook</code> methods could support a new kwarg to set whether it should commit at the end.</p>
<p dir="auto">Thoughts?</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="def _execute(self):
        command = self.task_instance.command(
            raw=True,
            ignore_dependencies=self.ignore_dependencies,
            force=self.force,
            pickle_id=self.pickle_id,
            mark_success=self.mark_success,
            task_start_date=self.task_start_date,
            job_id=self.id,
            pool=self.pool,
        )
        self.process = subprocess.Popen(['bash', '-c', command])
        return_code = None
        while return_code is None:
            self.heartbeat()
            return_code = self.process.poll()"><pre class="notranslate"><code class="notranslate">def _execute(self):
        command = self.task_instance.command(
            raw=True,
            ignore_dependencies=self.ignore_dependencies,
            force=self.force,
            pickle_id=self.pickle_id,
            mark_success=self.mark_success,
            task_start_date=self.task_start_date,
            job_id=self.id,
            pool=self.pool,
        )
        self.process = subprocess.Popen(['bash', '-c', command])
        return_code = None
        while return_code is None:
            self.heartbeat()
            return_code = self.process.poll()
</code></pre></div>
<p dir="auto">Here we need to check the return_code of the process, otherwise it's possible that a process is killed externally but LocalTaskJob think it suceeds</p>