Fail explicitly in storage when asking for duplicate keys (#530)

When debugging #529 I realized this is a silent problem (we might access
memory of a vector that is not allocated), so we should just explicitly
fail.

modyn/storage/include/internal/grpc/storage_service_impl.hpp

@@ -571,6 +571,19 @@ class StorageServiceImpl final : public modyn::storage::Storage::Service {
       session << sample_query, soci::into(sample_labels), soci::into(sample_indices), soci::into(sample_fileids),
           soci::use(dataset_data.dataset_id);
 
+      if (sample_fileids.size() != num_keys) {
+        SPDLOG_ERROR(fmt::format("Sample query is {}", sample_query));
+        SPDLOG_ERROR(
+            fmt::format("num_keys = {}\n sample_labels = [{}]\n sample_indices = [{}]\n "
+                        "sample_fileids = [{}]",
+                        num_keys, fmt::join(sample_labels, ", "), fmt::join(sample_indices, ", "),
+                        fmt::join(sample_fileids, ", ")));
+        throw modyn::utils::ModynException(
+            fmt::format("Got back {} samples from DB, while asking for {} keys. You might have asked for duplicate "
+                        "keys, which is not supported.",
+                        sample_fileids.size(), num_keys));
+      }
+
       int64_t current_file_id = sample_fileids.at(0);
       uint64_t current_file_start_idx = 0;
       std::string current_file_path;
@@ -579,6 +592,11 @@ class StorageServiceImpl final : public modyn::storage::Storage::Service {
 
       if (current_file_path.empty() || current_file_path.find_first_not_of(' ') == std::string::npos) {
         SPDLOG_ERROR(fmt::format("Sample query is {}", sample_query));
+        SPDLOG_ERROR(
+            fmt::format("num_keys = {}, current_file_id = {}\n sample_labels = [{}]\n sample_indices = [{}]\n "
+                        "sample_fileids = [{}]",
+                        num_keys, current_file_id, fmt::join(sample_labels, ", "), fmt::join(sample_indices, ", "),
+                        fmt::join(sample_fileids, ", ")));
         throw modyn::utils::ModynException(fmt::format("Could not obtain full path of file id {} in dataset {}",
                                                        current_file_id, dataset_data.dataset_id));
       }
@@ -591,7 +609,7 @@ class StorageServiceImpl final : public modyn::storage::Storage::Service {
                            file_wrapper_config_node, filesystem_wrapper);
 
       for (uint64_t sample_idx = 0; sample_idx < num_keys; ++sample_idx) {
-        const int64_t& sample_fileid = sample_fileids[sample_idx];
+        const int64_t& sample_fileid = sample_fileids.at(sample_idx);
 
         if (sample_fileid != current_file_id) {
           // 1. Prepare response
@@ -629,6 +647,12 @@ class StorageServiceImpl final : public modyn::storage::Storage::Service {
           soci::into(current_file_path), soci::use(current_file_id), soci::use(dataset_data.dataset_id);
           if (current_file_path.empty() || current_file_path.find_first_not_of(' ') == std::string::npos) {
             SPDLOG_ERROR(fmt::format("Sample query is {}", sample_query));
+            const int64_t& previous_fid = sample_fileids.at(sample_idx - 1);
+            SPDLOG_ERROR(
+                fmt::format("num_keys = {}, sample_idx = {}, previous_fid = {}\n sample_labels = [{}]\n sample_indices "
+                            "= [{}]\n sample_fileids = [{}]",
+                            num_keys, sample_idx, previous_fid, fmt::join(sample_labels, ", "),
+                            fmt::join(sample_indices, ", "), fmt::join(sample_fileids, ", ")));
             throw modyn::utils::ModynException(fmt::format("Could not obtain full path of file id {} in dataset {}",
                                                            current_file_id, dataset_data.dataset_id));
           }