Don't preprocess a storage file more than once (#895)

* Don't preprocess a storage file more than once

Fixes problem on file decompression, which would try to decompress the already-decompressed file

Signed-off-by: Sonia Jin <diegothemuich@gmail.com>
Co-authored-by: Emerson Knapp <emerson.b.knapp@gmail.com>

rosbag2_compression/test/rosbag2_compression/test_sequential_compression_reader.cpp

@@ -284,3 +284,26 @@ TEST_F(SequentialCompressionReaderTest, can_find_prefixed_filenames_in_renamed_b
   EXPECT_NO_THROW(reader->open(storage_options_, converter_options_));
   EXPECT_TRUE(reader->has_next_file());
 }
+
+TEST_F(SequentialCompressionReaderTest, does_not_decompress_again_on_seek)
+{
+  auto decompressor = std::make_unique<NiceMock<MockDecompressor>>();
+  ON_CALL(*decompressor, decompress_uri(_)).WillByDefault(Return("some/path"));
+  EXPECT_CALL(*decompressor, decompress_uri(_)).Times(1);
+
+  auto compression_factory = std::make_unique<NiceMock<MockCompressionFactory>>();
+  ON_CALL(*compression_factory, create_decompressor(_))
+  .WillByDefault(Return(ByMove(std::move(decompressor))));
+
+  ON_CALL(*storage_, has_next()).WillByDefault(Return(true));
+
+  auto sequential_reader = std::make_unique<rosbag2_compression::SequentialCompressionReader>(
+    std::move(compression_factory),
+    std::move(storage_factory_),
+    converter_factory_,
+    std::move(metadata_io_));
+
+  reader_ = std::make_unique<rosbag2_cpp::Reader>(std::move(sequential_reader));
+  reader_->open(storage_options_, converter_options_);
+  reader_->seek(0);
+}

rosbag2_cpp/include/rosbag2_cpp/readers/sequential_reader.hpp

@@ -17,6 +17,7 @@
 
 #include <memory>
 #include <string>
+#include <unordered_set>
 #include <vector>
 
 #include "rosbag2_cpp/converter.hpp"
@@ -147,7 +148,9 @@ class ROSBAG2_CPP_PUBLIC SequentialReader
 
   /**
     * Prepare current file for opening by the storage implementation.
-    * This may be used by subclasses, for example decompressing
+    * This may be used by subclasses, for example decompressing.
+    * This should be a once-per-file operation, meaning that subsequent opening
+    * of the same file will not trigger another preprocessing.
     */
   virtual void preprocess_current_file() {}
 
@@ -161,6 +164,7 @@ class ROSBAG2_CPP_PUBLIC SequentialReader
   std::vector<rosbag2_storage::TopicMetadata> topics_metadata_{};
   std::vector<std::string> file_paths_{};  // List of database files.
   std::vector<std::string>::iterator current_file_iterator_{};  // Index of file to read from
+  std::unordered_set<std::string> preprocessed_file_paths_;  // List of preprocessed paths
 
   // Hang on to this because storage_options_ is mutated to point at individual files
   std::string base_folder_;

rosbag2_cpp/src/rosbag2_cpp/readers/sequential_reader.cpp

@@ -95,18 +95,10 @@ void SequentialReader::open(
       ROSBAG2_CPP_LOG_WARN("No file paths were found in metadata.");
       return;
     }
-
     file_paths_ = details::resolve_relative_paths(
       storage_options.uri, metadata_.relative_file_paths, metadata_.version);
     current_file_iterator_ = file_paths_.begin();
-
-    preprocess_current_file();
-
-    storage_options_.uri = get_current_file();
-    storage_ = storage_factory_->open_read_only(storage_options_);
-    if (!storage_) {
-      throw std::runtime_error{"No storage could be initialized. Abort"};
-    }
+    load_current_file();
   } else {
     storage_ = storage_factory_->open_read_only(storage_options_);
     if (!storage_) {
@@ -215,10 +207,19 @@ bool SequentialReader::has_next_file() const
 
 void SequentialReader::load_current_file()
 {
-  preprocess_current_file();
+  // only preprocess if file hasn't been preprocessed before
+  // add path AFTER preprocessing since preprocessing may modify it
+  if (preprocessed_file_paths_.find(get_current_file()) == preprocessed_file_paths_.end()) {
+    preprocess_current_file();
+    preprocessed_file_paths_.insert(get_current_file());
+  }
+  // open and check storage exists
   storage_options_.uri = get_current_file();
-  // open and set filters
   storage_ = storage_factory_->open_read_only(storage_options_);
+  if (!storage_) {
+    throw std::runtime_error{"No storage could be initialized. Abort"};
+  }
+  // set filters
   storage_->seek(seek_time_);
   set_filter(topics_filter_);
 }

rosbag2_cpp/test/rosbag2_cpp/test_sequential_reader.cpp

@@ -140,7 +140,8 @@ TEST_F(SequentialReaderTest, set_filter_calls_storage) {
   EXPECT_ANY_THROW(reader_->get_implementation_handle().set_filter(storage_filter));
   EXPECT_ANY_THROW(reader_->get_implementation_handle().reset_filter());
 
-  EXPECT_CALL(*storage_, set_filter(_)).Times(3);
+  // Three times + initial open
+  EXPECT_CALL(*storage_, set_filter(_)).Times(4);
   reader_->open(default_storage_options_, {"", storage_serialization_format_});
   reader_->get_implementation_handle().set_filter(storage_filter);
   reader_->read_next();