Merge f670aec into 0a53ed1

Author: ivanmorozov333

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -1406,6 +1406,7 @@ class TTxAskPortionChunks: public TTransactionBase<TColumnShard> {
     THashMap<ui64, std::vector<NOlap::TPortionInfo::TConstPtr>> PortionsByPath;
     std::vector<TPortionConstructorV2> FetchedAccessors;
     const TString Consumer;
+    THashMap<NOlap::TPortionAddress, TPortionConstructorV2> Constructors;
 
 public:
     TTxAskPortionChunks(TColumnShard* self, const std::shared_ptr<NOlap::NDataAccessorControl::IAccessorCallback>& fetchCallback,
@@ -1428,13 +1429,42 @@ class TTxAskPortionChunks: public TTransactionBase<TColumnShard> {
         for (auto&& i : PortionsByPath) {
             AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("size", i.second.size())("path_id", i.first);
             for (auto&& p : i.second) {
+                auto itPortionConstructor = Constructors.find(p->GetAddress());
+                if (itPortionConstructor == Constructors.end()) {
+                    auto rowset = db.Table<NColumnShard::Schema::IndexColumnsV2>().Key(p->GetPathId(), p->GetPortionId()).Select();
+                    if (!rowset.IsReady()) {
+                        reask = true;
+                    } else {
+                        TPortionConstructorV2 constructor(p);
+                        AFL_VERIFY(!rowset.EndOfSet())("path_id", p->GetPathId())("portion_id", p->GetPortionId())(
+                            "debug", p->DebugString(true));
+                        NOlap::TColumnChunkLoadContextV2 info(rowset);
+                        constructor.SetRecords(std::move(info));
+                        itPortionConstructor = Constructors.emplace(p->GetAddress(), std::move(constructor)).first;
+                    }
+                }
                 if (!p->GetSchema(Self->GetIndexAs<NOlap::TColumnEngineForLogs>().GetVersionedIndex())->GetIndexesCount()) {
                     continue;
                 }
                 {
                     auto rowset = db.Table<NColumnShard::Schema::IndexIndexes>().Prefix(p->GetPathId(), p->GetPortionId()).Select();
                     if (!rowset.IsReady()) {
                         reask = true;
+                    } else {
+                        std::vector<NOlap::TIndexChunkLoadContext> indexes;
+                        bool localReask = false;
+                        while (!localReask && !rowset.EndOfSet()) {
+                            indexes.emplace_back(NOlap::TIndexChunkLoadContext(rowset, &selector));
+                            if (!rowset.Next()) {
+                                reask = true;
+                                localReask = true;
+                            }
+                        }
+                        if (!localReask) {
+                            itPortionConstructor->second.SetIndexes(std::move(indexes));
+                            FetchedAccessors.emplace_back(std::move(itPortionConstructor->second));
+                            Constructors.erase(itPortionConstructor);
+                        }
                     }
                 }
             }
@@ -1443,42 +1473,6 @@ class TTxAskPortionChunks: public TTransactionBase<TColumnShard> {
             return false;
         }
 
-        for (auto&& i : PortionsByPath) {
-            AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("stage", "processing")("size", i.second.size())(
-                "path_id", i.first);
-            while (i.second.size()) {
-                auto p = i.second.back();
-                TPortionConstructorV2 constructor(p);
-                {
-                    auto rowset = db.Table<NColumnShard::Schema::IndexColumnsV2>().Key(p->GetPathId(), p->GetPortionId()).Select();
-                    if (!rowset.IsReady()) {
-                        return false;
-                    }
-                    AFL_VERIFY(!rowset.EndOfSet())("path_id", p->GetPathId())("portion_id", p->GetPortionId())("debug", p->DebugString(true));
-                    NOlap::TColumnChunkLoadContextV2 info(rowset);
-                    constructor.SetRecords(std::move(info));
-                }
-                std::vector<NOlap::TIndexChunkLoadContext> indexes;
-                if (p->GetSchema(Self->GetIndexAs<NOlap::TColumnEngineForLogs>().GetVersionedIndex())->GetIndexesCount()) {
-                    auto rowset = db.Table<NColumnShard::Schema::IndexIndexes>().Prefix(p->GetPathId(), p->GetPortionId()).Select();
-                    if (!rowset.IsReady()) {
-                        return false;
-                    }
-                    while (!rowset.EndOfSet()) {
-                        indexes.emplace_back(NOlap::TIndexChunkLoadContext(rowset, &selector));
-                        if (!rowset.Next()) {
-                            return false;
-                        }
-                    }
-                }
-                constructor.SetIndexes(std::move(indexes));
-                FetchedAccessors.emplace_back(std::move(constructor));
-                i.second.pop_back();
-            }
-            AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("stage", "finished")("size", i.second.size())(
-                "path_id", i.first);
-        }
-
         AFL_INFO(NKikimrServices::TX_COLUMNSHARD)("stage", "finished");
         NConveyor::TInsertServiceOperator::AsyncTaskToExecute(std::make_shared<TAccessorsParsingTask>(FetchCallback, std::move(FetchedAccessors)));
         return true;