data_dictionary: compose the location with "/"

in 787ea4b1, we construct a new `storage_options` for each sstable
to be restored. the `location` of the new `storage_option` instances
is composed of the configured `prefix` and the dirname of each toc
component. but instead of separating them with "/", we just concatenate
them. this breaks the test if the specified key representing toc
components includes "dirname" in them.

in this change

- data_directory: instead of using "{prefix}{dirname}", we use
  "{prefix}/{dirname}".
- test/object_store: update the existing test to add a suffix
  in the keys of the toc objects to mimic the typical use case.

Signed-off-by: Kefu Chai <kefu.chai@scylladb.com>

Closes scylladb/scylladb#21170

data_dictionary/data_dictionary.cc

@@ -362,10 +362,16 @@ storage_options storage_options::append_to_s3_prefix(const sstring& s) const {
     //     desc: me-3gdq_0bki_2cvk02wubgncy8qd41-big
     SCYLLA_ASSERT(!is_local_type());
     storage_options ret = *this;
+    if (s.empty()) {
+        // scylla-manager should always pass sstables with non-empty dirname,
+        // but still..
+        return ret;
+    }
+
     s3 s3_options = std::get<s3>(value);
     SCYLLA_ASSERT(std::holds_alternative<sstring>(s3_options.location));
     sstring prefix = std::get<sstring>(s3_options.location);
-    s3_options.location = prefix + s;
+    s3_options.location = seastar::format("{}/{}", prefix, s);
     ret.value = std::move(s3_options);
     return ret;
 }

test/object_store/test_backup.py

@@ -153,10 +153,30 @@ async def test_simple_backup_and_restore(manager: ManagerClient, s3_server):
     orig_res = cql.execute(f"SELECT * FROM {ks}.{cf}")
     orig_rows = { x.name: x.value for x in orig_res }
 
-    toc_names = [entry.name for entry in list_sstables() if entry.name.endswith('TOC.txt')]
+    # include a "suffix" in the key to mimic the use case where scylla-manager
+    # 1. backups sstables of multiple snapshots, and deduplicate the backup'ed
+    #    sstables by only upload the new sstables
+    # 2. restore a given snapshot by collecting all sstables of this snapshot from
+    #    multiple places
+    #
+    # in this test, we:
+    # 1. upload:
+    #    prefix: {prefix}/{suffix}
+    #    sstables:
+    #    - 1-TOC.txt
+    #    - 2-TOC.txt
+    #    - ...
+    # 2. download:
+    #    prefix = {prefix}
+    #    sstables:
+    #    - {suffix}/1-TOC.txt
+    #    - {suffix}/2-TOC.txt
+    #    - ...
+    suffix = 'suffix'
+    toc_names = [f'{suffix}/{entry.name}' for entry in list_sstables() if entry.name.endswith('TOC.txt')]
 
     prefix = f'{cf}/{snap_name}'
-    tid = await manager.api.backup(server.ip_addr, ks, cf, snap_name, s3_server.address, s3_server.bucket_name, prefix)
+    tid = await manager.api.backup(server.ip_addr, ks, cf, snap_name, s3_server.address, s3_server.bucket_name, f'{prefix}/{suffix}')
     status = await manager.api.wait_task(server.ip_addr, tid)
     assert (status is not None) and (status['state'] == 'done')