Merge pull request #35922 from kssenii/fix-race-in-cached-buffer

Fix race in cached buffer

src/Disks/IO/CachedReadBufferFromRemoteFS.cpp

@@ -334,15 +334,17 @@ SeekableReadBufferPtr CachedReadBufferFromRemoteFS::getImplementationBuffer(File
                 read_buffer_for_file_segment->seek(file_offset_of_buffer_end, SEEK_SET);
             }
 
-            auto impl_range = read_buffer_for_file_segment->getRemainingReadRange();
             auto download_offset = file_segment->getDownloadOffset();
             if (download_offset != static_cast<size_t>(read_buffer_for_file_segment->getPosition()))
+            {
+                auto impl_range = read_buffer_for_file_segment->getRemainingReadRange();
                 throw Exception(
                     ErrorCodes::LOGICAL_ERROR,
                     "Buffer's offsets mismatch; cached buffer offset: {}, download_offset: {}, position: {}, implementation buffer offset: {}, "
                     "implementation buffer reading until: {}, file segment info: {}",
                     file_offset_of_buffer_end, download_offset, read_buffer_for_file_segment->getPosition(),
                     impl_range.left, *impl_range.right, file_segment->getInfoForLog());
+            }
 
             break;
         }
@@ -802,12 +804,14 @@ std::optional<size_t> CachedReadBufferFromRemoteFS::getLastNonDownloadedOffset()
 
 String CachedReadBufferFromRemoteFS::getInfoForLog()
 {
-    auto implementation_buffer_read_range_str =
+    String implementation_buffer_read_range_str;
-        implementation_buffer ?
+    if (implementation_buffer)
-        std::to_string(implementation_buffer->getRemainingReadRange().left)
+    {
-        + '-'
+        auto read_range = implementation_buffer->getRemainingReadRange();
-        + (implementation_buffer->getRemainingReadRange().right ? std::to_string(*implementation_buffer->getRemainingReadRange().right) : "None")
+        implementation_buffer_read_range_str = std::to_string(read_range.left) + '-' + (read_range.right ? std::to_string(*read_range.right) : "None");
-        : "None";
+    }
+    else
+        implementation_buffer_read_range_str = "None";
 
     auto current_file_segment_info = current_file_segment_it == file_segments_holder->file_segments.end() ? "None" : (*current_file_segment_it)->getInfoForLog();
 

src/IO/ReadBufferFromS3.cpp

@@ -240,7 +240,7 @@ void ReadBufferFromS3::setReadUntilPosition(size_t position)
 
 SeekableReadBuffer::Range ReadBufferFromS3::getRemainingReadRange() const
 {
-    return Range{.left = static_cast<size_t>(offset), .right = read_until_position ? std::optional{read_until_position - 1} : std::nullopt};
+    return Range{ .left = static_cast<size_t>(offset), .right = read_until_position ? std::optional{read_until_position - 1} : std::nullopt };
 }
 
 std::unique_ptr<ReadBuffer> ReadBufferFromS3::initialize()

src/IO/ReadBufferFromS3.h

@@ -33,8 +33,11 @@ private:
     String key;
     UInt64 max_single_read_retries;
 
-    off_t offset = 0;
+    /// These variables are atomic because they can be used for `logging only`
-    off_t read_until_position = 0;
+    /// (where it is not important to get consistent result)
+    /// from separate thread other than the one which uses the buffer for s3 reading.
+    std::atomic<off_t> offset = 0;
+    std::atomic<off_t> read_until_position = 0;
 
     Aws::S3::Model::GetObjectResult read_result;
     std::unique_ptr<ReadBuffer> impl;