Merge pull request ClickHouse#39531 from ClickHouse/backport/22.3/36763

Backport ClickHouse#36763 to 22.3: Fix some issues with low cardinality and async reads from remote fs

Author: kssenii

src/Disks/IO/AsynchronousReadIndirectBufferFromRemoteFS.cpp

@@ -134,7 +134,10 @@ void AsynchronousReadIndirectBufferFromRemoteFS::prefetch()
 void AsynchronousReadIndirectBufferFromRemoteFS::setReadUntilPosition(size_t position)
 {
     if (prefetch_future.valid())
-        throw Exception(ErrorCodes::LOGICAL_ERROR, "Prefetch is valid in readUntilPosition");
+    {
+        prefetch_future.wait();
+        prefetch_future = {};
+    }
 
     if (position > read_until_position)
     {
@@ -147,7 +150,10 @@ void AsynchronousReadIndirectBufferFromRemoteFS::setReadUntilPosition(size_t pos
 void AsynchronousReadIndirectBufferFromRemoteFS::setReadUntilEnd()
 {
     if (prefetch_future.valid())
-        throw Exception(ErrorCodes::LOGICAL_ERROR, "Prefetch is valid in readUntilEnd");
+    {
+        prefetch_future.wait();
+        prefetch_future = {};
+    }
 
     read_until_position = impl->getFileSize();
     impl->setReadUntilPosition(*read_until_position);

src/Disks/IO/ReadBufferFromRemoteFSGather.cpp

@@ -271,7 +271,7 @@ String ReadBufferFromRemoteFSGather::getInfoForLog()
 size_t ReadBufferFromRemoteFSGather::getImplementationBufferOffset() const
 {
     if (!current_buf)
-        throw Exception(ErrorCodes::LOGICAL_ERROR, "Buffer not initialized");
+        return file_offset_of_buffer_end;
 
     return current_buf->getFileOffsetOfBufferEnd();
 }

tests/queries/1_stateful/00167_read_bytes_from_fs.sql

@@ -1,3 +1,5 @@
+-- Tags: no-random-settings
+
 SELECT sum(cityHash64(*)) FROM test.hits SETTINGS max_threads=40;
 
 -- We had a bug which lead to additional compressed data read. test.hits compressed size is about 1.2Gb, but we read more then 3Gb.