support fan out for backfill read and fix more

Author: yl-lisen

src/Storages/Streaming/StorageStream.cpp

@@ -486,15 +486,19 @@ void StorageStream::readConcat(
     for (auto & stream_shard : shards_to_read)
     {
         auto create_streaming_source = [this, header, storage_snapshot, stream_shard, seek_to_info = query_info.seek_to_info, context_](
-                                           Int64 & max_sn_in_parts) {
+                                           Int64 & max_sn_in_parts) -> SourcePtr {
             if (max_sn_in_parts < 0)
             {
                 /// Fallback to seek streaming store
                 auto offsets = stream_shard->getOffsets(seek_to_info);
                 LOG_INFO(log, "Fused read fallbacks to seek stream for shard={} since there are no historical data", stream_shard->shard);
 
-                return std::make_shared<StreamingStoreSource>(
-                    stream_shard, header, storage_snapshot, context_, offsets[stream_shard->shard], log);
+                if (context_->getSettingsRef().query_resource_group.value == "shared")
+                    return source_multiplexers->createChannel(
+                        stream_shard, header.getNames(), storage_snapshot, context_, offsets[stream_shard->shard]);
+                else
+                    return std::make_shared<StreamingStoreSource>(
+                        stream_shard, header, storage_snapshot, context_, offsets[stream_shard->shard], log);
             }
 
             auto committed = stream_shard->storage->inMemoryCommittedSN();
@@ -526,7 +530,12 @@ void StorageStream::readConcat(
                     max_sn_in_parts,
                     committed);
 
-                return std::make_shared<StreamingStoreSource>(stream_shard, header, storage_snapshot, context_, max_sn_in_parts + 1, log);
+                if (context_->getSettingsRef().query_resource_group.value == "shared")
+                    return source_multiplexers->createChannel(
+                        stream_shard, header.getNames(), storage_snapshot, context_, max_sn_in_parts + 1);
+                else
+                    return std::make_shared<StreamingStoreSource>(
+                        stream_shard, header, storage_snapshot, context_, max_sn_in_parts + 1, log);
             }
             else
             {
@@ -542,8 +551,12 @@ void StorageStream::readConcat(
 
                 /// We need reset max_sn_in_parts to tell caller that we are seeking streaming store directly
                 max_sn_in_parts = -1;
-                return std::make_shared<StreamingStoreSource>(
-                    stream_shard, header, storage_snapshot, context_, offsets[stream_shard->shard], log);
+                if (context_->getSettingsRef().query_resource_group.value == "shared")
+                    return source_multiplexers->createChannel(
+                        stream_shard, header.getNames(), storage_snapshot, context_, offsets[stream_shard->shard]);
+                else
+                    return std::make_shared<StreamingStoreSource>(
+                        stream_shard, header, storage_snapshot, context_, offsets[stream_shard->shard], log);
             }
         };
 

src/Storages/Streaming/StreamingStoreSourceMultiplexer.cpp

@@ -280,6 +280,8 @@ std::pair<String, Int32> StreamingStoreSourceMultiplexer::getStreamShard() const
     return stream_shard->getStreamShard();
 }
 
+std::atomic<uint32_t> StreamingStoreSourceMultiplexers::multiplexer_id = 0;
+
 StreamingStoreSourceMultiplexers::StreamingStoreSourceMultiplexers(ContextPtr global_context_, Poco::Logger * log_)
     : global_context(std::move(global_context_)), log(log_)
 {
@@ -310,7 +312,8 @@ StreamingStoreSourceChannelPtr StreamingStoreSourceMultiplexers::createChannel(
     {
         multiplexers.emplace(
             shard,
-            StreamingStoreSourceMultiplexerPtrs{std::make_shared<StreamingStoreSourceMultiplexer>(0, stream_shard, global_context, log)});
+            StreamingStoreSourceMultiplexerPtrs{
+                std::make_shared<StreamingStoreSourceMultiplexer>(getMultiplexerID(), stream_shard, global_context, log)});
         iter = multiplexers.find(shard);
     }
 
@@ -342,7 +345,7 @@ StreamingStoreSourceChannelPtr StreamingStoreSourceMultiplexers::createChannel(
         /// If min channels is greater than > 20(default value), create another multiplexer for this shard
         if (min_channels > global_context->getSettingsRef().max_channels_per_resource_group.value)
         {
-            best_multiplexer = std::make_shared<StreamingStoreSourceMultiplexer>(iter->second.size(), stream_shard, global_context, log);
+            best_multiplexer = std::make_shared<StreamingStoreSourceMultiplexer>(getMultiplexerID(), stream_shard, global_context, log);
             iter->second.push_back(best_multiplexer);
         }
 
@@ -351,7 +354,7 @@ StreamingStoreSourceChannelPtr StreamingStoreSourceMultiplexers::createChannel(
     else
     {
         /// All multiplexers are shutdown
-        auto multiplexer{std::make_shared<StreamingStoreSourceMultiplexer>(iter->second.size(), stream_shard, global_context, log)};
+        auto multiplexer{std::make_shared<StreamingStoreSourceMultiplexer>(getMultiplexerID(), stream_shard, global_context, log)};
         iter->second.push_back(multiplexer);
         return multiplexer->createChannel(column_names, storage_snapshot, query_context);
     }
@@ -366,7 +369,7 @@ StreamingStoreSourceChannelPtr StreamingStoreSourceMultiplexers::createIndepende
     /// will startup after `StreamingStoreSourceChannel::recover()` and reset recovered sn
     /// The `multiplexer` is cached in created StreamingStoreSourceChannel, we can release this one
     auto multiplexer = std::make_shared<StreamingStoreSourceMultiplexer>(
-        0, std::move(stream_shard), global_context, log, [this](auto multiplexer_) { attachToSharedGroup(multiplexer_); });
+        getMultiplexerID(), std::move(stream_shard), global_context, log, [this](auto multiplexer_) { attachToSharedGroup(multiplexer_); });
     return multiplexer->createChannel(column_names, storage_snapshot, query_context);
 }
 
@@ -379,7 +382,7 @@ StreamingStoreSourceChannelPtr StreamingStoreSourceMultiplexers::createIndepende
 {
     /// The `multiplexer` is cached in created StreamingStoreSourceChannel, we can release this one
     auto multiplexer = std::make_shared<StreamingStoreSourceMultiplexer>(
-        0, std::move(stream_shard), global_context, log, [this](auto multiplexer_) { attachToSharedGroup(multiplexer_); });
+        getMultiplexerID(), std::move(stream_shard), global_context, log, [this](auto multiplexer_) { attachToSharedGroup(multiplexer_); });
     auto channel = multiplexer->createChannel(column_names, storage_snapshot, query_context);
     multiplexer->resetSequenceNumber(start_sn);
     multiplexer->startup();
@@ -394,8 +397,17 @@ void StreamingStoreSourceMultiplexers::attachToSharedGroup(StreamingStoreSourceM
     detached_multiplexers.clear();
 
     auto & multiplexer_list = multiplexers[multiplexer->stream_shard->getShard()];
-    for (auto & shared_multiplexer : multiplexer_list)
+    for (auto it = multiplexer_list.begin(); it != multiplexer_list.end();)
     {
+        if ((*it)->isShutdown())
+        {
+            it = multiplexer_list.erase(it);
+            continue;
+        }
+
+        auto & shared_multiplexer = *it;
+        ++it;
+
         /// Skip multiplexer that already have too many channels
         if (shared_multiplexer->totalChannels() > global_context->getSettingsRef().max_channels_per_resource_group.value)
             continue;
@@ -410,7 +422,6 @@ void StreamingStoreSourceMultiplexers::attachToSharedGroup(StreamingStoreSourceM
     }
 
     /// Not detach channels into any existed shared multiplexer, so we reuse it and join in shared groups
-    multiplexer->id = multiplexer_list.size(); /// it's thread safe
     multiplexer_list.emplace_back(std::move(multiplexer));
 }
 }

src/Storages/Streaming/StreamingStoreSourceMultiplexer.h

@@ -122,10 +122,14 @@ class StreamingStoreSourceMultiplexers final
 
     void attachToSharedGroup(StreamingStoreSourceMultiplexerPtr multiplexer);
 
+    uint32_t getMultiplexerID() { return multiplexer_id.fetch_add(1); }
+
 private:
     ContextPtr global_context;
     Poco::Logger * log;
 
+    static std::atomic<uint32_t> multiplexer_id;
+
     std::mutex multiplexers_mutex;
     std::unordered_map<Int32, StreamingStoreSourceMultiplexerPtrs> multiplexers;
     StreamingStoreSourceMultiplexerPtrs detached_multiplexers;