dbsp: Add worker thread index to file names.

I think that this will make troubleshooting easier sometimes.

I'd like to add information about the operator in use, too, but it's harder
to get that information.

Signed-off-by: Ben Pfaff <blp@feldera.com>

crates/dbsp/src/storage/backend/mod.rs

@@ -190,8 +190,14 @@ pub trait Storage {
     /// called and the [`FileHandle`] is converted to an
     /// [`ImmutableFileHandle`].
     fn create(&self) -> Result<FileHandle, StorageError> {
+        self.create_with_prefix("")
+    }
+
+    /// Creates a new persistent file used for writing data, giving the file's
+    /// name the specified `prefix`. See also [`create`](Self::create).
+    fn create_with_prefix(&self, prefix: &str) -> Result<FileHandle, StorageError> {
         let uuid = Uuid::now_v7();
-        let name = uuid.to_string() + CREATE_FILE_EXTENSION;
+        let name = format!("{}{}{}", prefix, uuid, CREATE_FILE_EXTENSION);
         let name_path = Path::new(&name);
         self.create_named(name_path)
     }

crates/dbsp/src/storage/file/writer.rs

@@ -33,6 +33,7 @@ use crate::storage::{
 use crate::{
     dynamic::{DataTrait, DeserializeDyn, SerializeDyn},
     storage::file::ItemFactory,
+    Runtime,
 };
 
 use super::cache::FileCache;
@@ -927,8 +928,9 @@ impl Writer {
             .map(|(column, factories)| ColumnWriter::new(factories, &parameters, column))
             .collect();
         let finished_columns = Vec::with_capacity(n_columns);
+        let worker = format!("w{}-", Runtime::worker_index());
         let writer = Self {
-            writer: BlockWriter::new(writer, writer.create()?),
+            writer: BlockWriter::new(writer, writer.create_with_prefix(&worker)?),
             cws,
             finished_columns,
         };