[Analytics] Stream sstable eagerly when bulk writing to allow reclaiming local disk space

Currently, each bulk write executor only sends sstables after exhausting the input data (of the task). All produced sstables are staged locally, when executor local disk space is limited or the input data size is too large, there is a risk of running out of disk space.
The patch changes the streaming strategy to stream eagerly and remove the local files sooner.