Don't remove already existing file when doing partitioning

Author: stloyd

src/core/etl/src/Flow/ETL/Filesystem/FilesystemStreams.php

@@ -41,7 +41,7 @@ public function closeStreams(Path $path) : void
 
                     if ($this->saveMode === SaveMode::Overwrite) {
                         if ($fileStream->path()->partitions()->count()) {
-                            $partitionFilesPatter = \Flow\Filesystem\DSL\path($fileStream->path()->parentDirectory()->uri() . '/*', $fileStream->path()->options());
+                            $partitionFilesPatter = Path::from($fileStream->path()->parentDirectory()->uri() . '/' . $fileStream->path()->basename(), $fileStream->path()->options());
 
                             foreach ($fs->list($partitionFilesPatter) as $partitionFile) {
                                 if (\str_contains($partitionFile->path->path(), self::FLOW_TMP_FILE_PREFIX)) {
@@ -54,7 +54,7 @@ public function closeStreams(Path $path) : void
 
                         $fs->mv(
                             $fileStream->path(),
-                            \Flow\Filesystem\DSL\path(
+                            Path::from(
                                 \str_replace(self::FLOW_TMP_FILE_PREFIX, '', $fileStream->path()->uri()),
                                 $fileStream->path()->options()
                             )

src/core/etl/tests/Flow/ETL/Tests/Integration/Filesystem/FilesystemStreams/Partitioned/OverwriteModeTest.php

@@ -27,6 +27,7 @@ public function test_open_stream_for_existing_partition_with_existing_file() : v
         $this->setupFiles([
             __FUNCTION__ => [
                 'partition=value' => [
+                    'existing.txt' => 'file content',
                     'file.txt' => 'file content',
                 ],
             ],
@@ -39,10 +40,10 @@ public function test_open_stream_for_existing_partition_with_existing_file() : v
 
         $files = \iterator_to_array($this->fs()->list(path($file->parentDirectory()->path() . '/**/*.txt')));
 
-        self::assertCount(1, $files);
+        self::assertCount(2, $files);
 
-        self::assertStringStartsWith('file.txt', $files[0]->path->basename());
-        self::assertSame('new content', \file_get_contents($files[0]->path->path()));
+        self::assertStringStartsWith('file.txt', $files[1]->path->basename());
+        self::assertSame('new content', \file_get_contents($files[1]->path->path()));
     }
 
     public function test_open_stream_for_existing_partition_without_existing_file() : void