Backport #78205 to 25.3: Fix prefetch of substreams with prefixes in Wide parts

robot-clickhouse · robot-clickhouse · commit aa2afaeda148 · 2025-04-01T13:17:09.000Z
diff --git a/src/Storages/MergeTree/MergeTreeReaderWide.cpp b/src/Storages/MergeTree/MergeTreeReaderWide.cpp
@@ -110,7 +110,7 @@ void MergeTreeReaderWide::prefetchForAllColumns(
         return;
 
     if (deserialize_prefixes)
-        deserializePrefixForAllColumnsWithPrefetch(num_columns, current_task_last_mark, priority);
+        deserializePrefixForAllColumnsWithPrefetch(num_columns, from_mark, current_task_last_mark, priority);
 
     /// Request reading of data in advance,
     /// so if reading can be asynchronous, it will also be performed in parallel for all columns.
@@ -151,7 +151,7 @@ size_t MergeTreeReaderWide::readRows(
             return max_rows_to_read;
 
         prefetchForAllColumns(Priority{}, num_columns, from_mark, current_task_last_mark, continue_reading, /*deserialize_prefixes=*/ true);
-        deserializePrefixForAllColumns(num_columns, current_task_last_mark);
+        deserializePrefixForAllColumns(num_columns, from_mark, current_task_last_mark);
 
         for (size_t pos = 0; pos < num_columns; ++pos)
         {
@@ -337,6 +337,7 @@ ReadBuffer * MergeTreeReaderWide::getStream(
 void MergeTreeReaderWide::deserializePrefix(
     const SerializationPtr & serialization,
     const NameAndTypePair & name_and_type,
+    size_t from_mark,
     size_t current_task_last_mark,
     DeserializeBinaryBulkStateMap & deserialize_state_map,
     ISerialization::SubstreamsCache & cache,
@@ -352,6 +353,15 @@ void MergeTreeReaderWide::deserializePrefix(
         deserialize_settings.prefixes_deserialization_thread_pool = settings.use_prefixes_deserialization_thread_pool ? &getMergeTreePrefixesDeserializationThreadPool().get() : nullptr;
         deserialize_settings.getter = [&](const ISerialization::SubstreamPath & substream_path)
         {
+            auto stream_name = IMergeTreeDataPart::getStreamNameForColumn(name_and_type, substream_path, data_part_info_for_read->getChecksums());
+            /// This stream could be prefetched in prefetchBeginOfRange, but here we
+            /// have to seek the stream to the start of file to deserialize the prefix.
+            /// If we do not read from the first mark, we should remove this stream from
+            /// prefetched_streams to prefetch it again starting from the current mark
+            /// after prefix is deserialized.
+            if (stream_name && from_mark != 0)
+                prefetched_streams.erase(*stream_name);
+
             return getStream(/* seek_to_start = */true, substream_path, data_part_info_for_read->getChecksums(), name_and_type, 0, /* seek_to_mark = */false, current_task_last_mark, cache);
         };
         /// Add streams for newly discovered dynamic subcolumns to start async marks loading beforehand if needed.
@@ -369,7 +379,7 @@ void MergeTreeReaderWide::deserializePrefix(
     }
 }
 
-void MergeTreeReaderWide::deserializePrefixForAllColumnsImpl(size_t num_columns, size_t current_task_last_mark, StreamCallbackGetter prefixes_prefetch_callback_getter)
+void MergeTreeReaderWide::deserializePrefixForAllColumnsImpl(size_t num_columns, size_t from_mark, size_t current_task_last_mark, StreamCallbackGetter prefixes_prefetch_callback_getter)
 {
     /// Check if we already deserialized prefixes.
     if (!deserialize_binary_bulk_state_map.empty())
@@ -387,6 +397,7 @@ void MergeTreeReaderWide::deserializePrefixForAllColumnsImpl(size_t num_columns,
                 deserializePrefix(
                     serializations[pos],
                     columns_to_read[pos],
+                    from_mark,
                     current_task_last_mark,
                     deserialize_state_map,
                     cache,
@@ -411,12 +422,12 @@ void MergeTreeReaderWide::deserializePrefixForAllColumnsImpl(size_t num_columns,
         deserialize_binary_bulk_state_map = deserialize();
 }
 
-void MergeTreeReaderWide::deserializePrefixForAllColumns(size_t num_columns, size_t current_task_last_mark)
+void MergeTreeReaderWide::deserializePrefixForAllColumns(size_t num_columns, size_t from_mark, size_t current_task_last_mark)
 {
-    deserializePrefixForAllColumnsImpl(num_columns, current_task_last_mark, {});
+    deserializePrefixForAllColumnsImpl(num_columns, from_mark, current_task_last_mark, {});
 }
 
-void MergeTreeReaderWide::deserializePrefixForAllColumnsWithPrefetch(size_t num_columns, size_t current_task_last_mark, Priority priority)
+void MergeTreeReaderWide::deserializePrefixForAllColumnsWithPrefetch(size_t num_columns, size_t from_mark, size_t current_task_last_mark, Priority priority)
 {
     auto prefixes_prefetch_callback_getter = [&](const NameAndTypePair & name_and_type)
     {
@@ -434,7 +445,7 @@ void MergeTreeReaderWide::deserializePrefixForAllColumnsWithPrefetch(size_t num_
         };
     };
 
-    deserializePrefixForAllColumnsImpl(num_columns, current_task_last_mark, prefixes_prefetch_callback_getter);
+    deserializePrefixForAllColumnsImpl(num_columns, from_mark, current_task_last_mark, prefixes_prefetch_callback_getter);
 }
 
 void MergeTreeReaderWide::prefetchForColumn(
@@ -494,7 +505,7 @@ void MergeTreeReaderWide::readData(
     ISerialization::DeserializeBinaryBulkSettings deserialize_settings;
     deserialize_settings.avg_value_size_hint = avg_value_size_hint;
 
-    deserializePrefix(serialization, name_and_type, current_task_last_mark, deserialize_binary_bulk_state_map, cache, deserialize_states_cache, {});
+    deserializePrefix(serialization, name_and_type, from_mark, current_task_last_mark, deserialize_binary_bulk_state_map, cache, deserialize_states_cache, {});
 
     deserialize_settings.getter = [&](const ISerialization::SubstreamPath & substream_path)
     {
diff --git a/src/Storages/MergeTree/MergeTreeReaderWide.h b/src/Storages/MergeTree/MergeTreeReaderWide.h
@@ -90,17 +90,18 @@ class MergeTreeReaderWide : public IMergeTreeReader
     void deserializePrefix(
         const SerializationPtr & serialization,
         const NameAndTypePair & name_and_type,
+        size_t from_mark,
         size_t current_task_last_mark,
         DeserializeBinaryBulkStateMap & deserialize_state_map,
         ISerialization::SubstreamsCache & cache,
         ISerialization::SubstreamsDeserializeStatesCache & deserialize_states_cache,
         ISerialization::StreamCallback prefixes_prefetch_callback);
 
-    void deserializePrefixForAllColumns(size_t num_columns, size_t current_task_last_mark);
-    void deserializePrefixForAllColumnsWithPrefetch(size_t num_columns, size_t current_task_last_mark, Priority priority);
+    void deserializePrefixForAllColumns(size_t num_columns, size_t from_mark, size_t current_task_last_mark);
+    void deserializePrefixForAllColumnsWithPrefetch(size_t num_columns, size_t from_mark, size_t current_task_last_mark, Priority priority);
 
     using StreamCallbackGetter = std::function<ISerialization::StreamCallback(const NameAndTypePair &)>;
-    void deserializePrefixForAllColumnsImpl(size_t num_columns, size_t current_task_last_mark, StreamCallbackGetter prefixes_prefetch_callback_getter);
+    void deserializePrefixForAllColumnsImpl(size_t num_columns, size_t from_mark, size_t current_task_last_mark, StreamCallbackGetter prefixes_prefetch_callback_getter);
 
     std::unordered_map<String, ISerialization::SubstreamsCache> caches;
     std::unordered_map<String, ISerialization::SubstreamsDeserializeStatesCache> deserialize_states_caches;
