Skip to content

Commit 2cffc4b

Browse files
ivanmorozov333ivanmorozov333
committed
Memory control and search fix (ydb-platform#1649)
* memory control and program data fetching for select fixes * fix * add queue for ttl * fix test build
1 parent bed4f87 commit 2cffc4b

18 files changed

+140
-31
lines changed

‎ydb/core/protos/config.proto

+1
Original file line numberDiff line numberDiff line change
@@ -1489,6 +1489,7 @@ message TColumnShardConfig {
14891489
optional uint32 WritingBufferDurationMs = 8 [default = 0];
14901490
optional bool UseChunkedMergeOnCompaction = 9 [default = true];
14911491
optional uint64 CompactionMemoryLimit = 10 [default = 536870912];
1492+
optional uint64 TieringsMemoryLimit = 11 [default = 536870912];
14921493
}
14931494

14941495
message TSchemeShardConfig {

‎ydb/core/tablet/resource_broker.cpp

+12
Original file line numberDiff line numberDiff line change
@@ -1269,6 +1269,7 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
12691269
const ui64 KqpRmQueueCPU = 4;
12701270
const ui64 KqpRmQueueMemory = 10ULL << 30;
12711271

1272+
const ui64 CSTTLCompactionMemoryLimit = 1ULL << 30;
12721273
const ui64 CSInsertCompactionMemoryLimit = 1ULL << 30;
12731274
const ui64 CSGeneralCompactionMemoryLimit = 3ULL << 30;
12741275
const ui64 CSScanMemoryLimit = 3ULL << 30;
@@ -1314,6 +1315,12 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
13141315
queue->MutableLimit()->SetCpu(3);
13151316
queue->MutableLimit()->SetMemory(CSInsertCompactionMemoryLimit);
13161317

1318+
queue = config.AddQueues();
1319+
queue->SetName("queue_cs_ttl");
1320+
queue->SetWeight(100);
1321+
queue->MutableLimit()->SetCpu(3);
1322+
queue->MutableLimit()->SetMemory(CSTTLCompactionMemoryLimit);
1323+
13171324
queue = config.AddQueues();
13181325
queue->SetName("queue_cs_general");
13191326
queue->SetWeight(100);
@@ -1413,6 +1420,11 @@ NKikimrResourceBroker::TResourceBrokerConfig MakeDefaultConfig()
14131420
task->SetQueueName("queue_compaction_borrowed");
14141421
task->SetDefaultDuration(TDuration::Minutes(10).GetValue());
14151422

1423+
task = config.AddTasks();
1424+
task->SetName("CS::TTL");
1425+
task->SetQueueName("queue_cs_ttl");
1426+
task->SetDefaultDuration(TDuration::Minutes(10).GetValue());
1427+
14161428
task = config.AddTasks();
14171429
task->SetName("CS::INDEXATION");
14181430
task->SetQueueName("queue_cs_indexation");

‎ydb/core/tx/columnshard/columnshard_impl.cpp

+8-3
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ TColumnShard::TColumnShard(TTabletStorageInfo* info, const TActorId& tablet)
8989
, SubscribeCounters(std::make_shared<NOlap::NResourceBroker::NSubscribe::TSubscriberCounters>())
9090
, InsertTaskSubscription(NOlap::TInsertColumnEngineChanges::StaticTypeName(), SubscribeCounters)
9191
, CompactTaskSubscription(NOlap::TCompactColumnEngineChanges::StaticTypeName(), SubscribeCounters)
92+
, TTLTaskSubscription(NOlap::TTTLColumnEngineChanges::StaticTypeName(), SubscribeCounters)
9293
, ReadCounters("Read")
9394
, ScanCounters("Scan")
9495
, WritesMonitor(*this)
@@ -743,21 +744,25 @@ bool TColumnShard::SetupTtl(const THashMap<ui64, NOlap::TTiering>& pathTtls, con
743744
}
744745

745746
auto actualIndexInfo = TablesManager.GetPrimaryIndex()->GetVersionedIndex();
746-
std::shared_ptr<NOlap::TTTLColumnEngineChanges> indexChanges = TablesManager.MutablePrimaryIndex().StartTtl(eviction, BackgroundController.GetConflictTTLPortions());
747+
const ui64 memoryUsageLimit = HasAppData() ? AppDataVerified().ColumnShardConfig.GetTieringsMemoryLimit() : ((ui64)512 * 1024 * 1024);
748+
std::shared_ptr<NOlap::TTTLColumnEngineChanges> indexChanges = TablesManager.MutablePrimaryIndex().StartTtl(
749+
eviction, BackgroundController.GetConflictTTLPortions(), memoryUsageLimit);
747750

748751
if (!indexChanges) {
749752
ACFL_DEBUG("background", "ttl")("skip_reason", "no_changes");
750753
return false;
751754
}
752-
755+
const TString externalTaskId = indexChanges->GetTaskIdentifier();
753756
const bool needWrites = indexChanges->NeedConstruction();
754757
ACFL_DEBUG("background", "ttl")("need_writes", needWrites);
755758

756759
indexChanges->Start(*this);
757760
auto ev = std::make_unique<TEvPrivate::TEvWriteIndex>(std::move(actualIndexInfo), indexChanges, false);
758761
NYDBTest::TControllers::GetColumnShardController()->OnWriteIndexStart(TabletID(), indexChanges->TypeString());
759762
if (needWrites) {
760-
ActorContext().Register(new NOlap::NBlobOperations::NRead::TActor(std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters)));
763+
NOlap::NResourceBroker::NSubscribe::ITask::StartResourceSubscription(
764+
ResourceSubscribeActor, std::make_shared<NOlap::NBlobOperations::NRead::ITask::TReadSubscriber>(
765+
std::make_shared<TChangesReadTask>(std::move(ev), SelfId(), TabletID(), CompactionCounters), 0, indexChanges->CalcMemoryForUsage(), externalTaskId, TTLTaskSubscription));
761766
} else {
762767
ev->SetPutStatus(NKikimrProto::OK);
763768
ActorContext().Send(SelfId(), std::move(ev));

‎ydb/core/tx/columnshard/columnshard_impl.h

+1
Original file line numberDiff line numberDiff line change
@@ -392,6 +392,7 @@ class TColumnShard
392392
std::shared_ptr<NOlap::NResourceBroker::NSubscribe::TSubscriberCounters> SubscribeCounters;
393393
NOlap::NResourceBroker::NSubscribe::TTaskContext InsertTaskSubscription;
394394
NOlap::NResourceBroker::NSubscribe::TTaskContext CompactTaskSubscription;
395+
NOlap::NResourceBroker::NSubscribe::TTaskContext TTLTaskSubscription;
395396
const TScanCounters ReadCounters;
396397
const TScanCounters ScanCounters;
397398
const TIndexationCounters CompactionCounters = TIndexationCounters("GeneralCompaction");

‎ydb/core/tx/columnshard/common/protos/snapshot.proto

+6
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
package NKikimrColumnShardProto;
2+
3+
message TSnapshot {
4+
optional uint64 PlanStep = 1;
5+
optional uint64 TxId = 2;
6+
}

‎ydb/core/tx/columnshard/common/protos/ya.make

+10
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
PROTO_LIBRARY()
2+
3+
SRCS(
4+
snapshot.proto
5+
)
6+
7+
PEERDIR(
8+
)
9+
10+
END()

‎ydb/core/tx/columnshard/common/snapshot.cpp

+7
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
#include "snapshot.h"
2+
#include <ydb/core/tx/columnshard/common/protos/snapshot.pb.h>
23
#include <util/string/builder.h>
34

45
namespace NKikimr::NOlap {
@@ -7,4 +8,10 @@ TString TSnapshot::DebugString() const {
78
return TStringBuilder() << "plan_step=" << PlanStep << ";tx_id=" << TxId << ";";
89
}
910

11+
NKikimrColumnShardProto::TSnapshot TSnapshot::SerializeToProto() const {
12+
NKikimrColumnShardProto::TSnapshot result;
13+
SerializeToProto(result);
14+
return result;
15+
}
16+
1017
};

‎ydb/core/tx/columnshard/common/snapshot.h

+26
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,11 @@
11
#pragma once
22
#include <util/stream/output.h>
33
#include <util/string/cast.h>
4+
#include <ydb/library/conclusion/status.h>
5+
6+
namespace NKikimrColumnShardProto {
7+
class TSnapshot;
8+
}
49

510
namespace NKikimr::NOlap {
611

@@ -47,6 +52,27 @@ class TSnapshot {
4752
return out << "{" << s.PlanStep << ':' << (s.TxId == std::numeric_limits<ui64>::max() ? "max" : ::ToString(s.TxId)) << "}";
4853
}
4954

55+
template <class TProto>
56+
void SerializeToProto(TProto& result) const {
57+
result.SetPlanStep(PlanStep);
58+
result.SetTxId(TxId);
59+
}
60+
61+
NKikimrColumnShardProto::TSnapshot SerializeToProto() const;
62+
63+
template <class TProto>
64+
TConclusionStatus DeserializeFromProto(const TProto& proto) {
65+
PlanStep = proto.GetPlanStep();
66+
TxId = proto.GetTxId();
67+
if (!PlanStep) {
68+
return TConclusionStatus::Fail("incorrect planStep in proto");
69+
}
70+
if (!TxId) {
71+
return TConclusionStatus::Fail("incorrect txId in proto");
72+
}
73+
return TConclusionStatus::Success();
74+
}
75+
5076
TString DebugString() const;
5177
};
5278

‎ydb/core/tx/columnshard/common/ya.make

+1
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ PEERDIR(
1111
ydb/core/protos
1212
contrib/libs/apache/arrow
1313
ydb/core/formats/arrow
14+
ydb/core/tx/columnshard/common/protos
1415
)
1516

1617
GENERATE_ENUM_SERIALIZATION(portion.h)

‎ydb/core/tx/columnshard/engines/changes/abstract/abstract.h

+6
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,12 @@ class TColumnEngineChanges {
173173
const TString TaskIdentifier = TGUID::Create().AsGuidString();
174174
virtual ui64 DoCalcMemoryForUsage() const = 0;
175175
public:
176+
class IMemoryPredictor {
177+
public:
178+
virtual ui64 AddPortion(const TPortionInfo& portionInfo) = 0;
179+
virtual ~IMemoryPredictor() = default;
180+
};
181+
176182
ui64 CalcMemoryForUsage() const {
177183
return DoCalcMemoryForUsage();
178184
}

‎ydb/core/tx/columnshard/engines/changes/general_compaction.h

-6
Original file line numberDiff line numberDiff line change
@@ -29,12 +29,6 @@ class TGeneralCompactColumnEngineChanges: public TCompactColumnEngineChanges {
2929
public:
3030
using TBase::TBase;
3131

32-
class IMemoryPredictor {
33-
public:
34-
virtual ui64 AddPortion(const TPortionInfo& portionInfo) = 0;
35-
virtual ~IMemoryPredictor() = default;
36-
};
37-
3832
class TMemoryPredictorSimplePolicy: public IMemoryPredictor {
3933
private:
4034
ui64 SumMemory = 0;

‎ydb/core/tx/columnshard/engines/changes/ttl.h

+20-1
Original file line numberDiff line numberDiff line change
@@ -51,13 +51,32 @@ class TTTLColumnEngineChanges: public TChangesWithAppend {
5151
virtual TConclusionStatus DoConstructBlobs(TConstructionContext& context) noexcept override;
5252
virtual NColumnShard::ECumulativeCounters GetCounterIndex(const bool isSuccess) const override;
5353
virtual ui64 DoCalcMemoryForUsage() const override {
54+
auto predictor = BuildMemoryPredictor();
5455
ui64 result = 0;
5556
for (auto& p : PortionsToEvict) {
56-
result += 2 * p.GetPortionInfo().GetBlobBytes();
57+
result = predictor->AddPortion(p.GetPortionInfo());
5758
}
5859
return result;
5960
}
6061
public:
62+
class TMemoryPredictorSimplePolicy: public IMemoryPredictor {
63+
private:
64+
ui64 SumBlobsMemory = 0;
65+
ui64 MaxRawMemory = 0;
66+
public:
67+
virtual ui64 AddPortion(const TPortionInfo& portionInfo) override {
68+
if (MaxRawMemory < portionInfo.GetRawBytes()) {
69+
MaxRawMemory = portionInfo.GetRawBytes();
70+
}
71+
SumBlobsMemory += portionInfo.GetBlobBytes();
72+
return SumBlobsMemory + MaxRawMemory;
73+
}
74+
};
75+
76+
static std::shared_ptr<IMemoryPredictor> BuildMemoryPredictor() {
77+
return std::make_shared<TMemoryPredictorSimplePolicy>();
78+
}
79+
6180
virtual bool NeedConstruction() const override {
6281
return PortionsToEvict.size();
6382
}

‎ydb/core/tx/columnshard/engines/column_engine.h

+2-2
Original file line numberDiff line numberDiff line change
@@ -375,8 +375,8 @@ class IColumnEngine {
375375
virtual std::shared_ptr<TColumnEngineChanges> StartCompaction(const TCompactionLimits& limits, const THashSet<TPortionAddress>& busyPortions) noexcept = 0;
376376
virtual std::shared_ptr<TCleanupColumnEngineChanges> StartCleanup(const TSnapshot& snapshot, THashSet<ui64>& pathsToDrop,
377377
ui32 maxRecords) noexcept = 0;
378-
virtual std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions,
379-
ui64 maxBytesToEvict = TCompactionLimits::DEFAULT_EVICTION_BYTES) noexcept = 0;
378+
virtual std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction,
379+
const THashSet<TPortionAddress>& busyPortions, const ui64 memoryUsageLimit) noexcept = 0;
380380
virtual bool ApplyChanges(IDbWrapper& db, std::shared_ptr<TColumnEngineChanges> changes, const TSnapshot& snapshot) noexcept = 0;
381381
virtual void RegisterSchemaVersion(const TSnapshot& snapshot, TIndexInfo&& info) = 0;
382382
virtual const TMap<ui64, std::shared_ptr<TColumnEngineStats>>& GetStats() const = 0;

‎ydb/core/tx/columnshard/engines/column_engine_logs.cpp

+10-10
Original file line numberDiff line numberDiff line change
@@ -312,7 +312,6 @@ std::shared_ptr<TCleanupColumnEngineChanges> TColumnEngineForLogs::StartCleanup(
312312

313313
TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering& ttl, TTieringProcessContext& context) const {
314314
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "ProcessTiering")("path_id", pathId)("ttl", ttl.GetDebugString());
315-
ui64 evictionSize = 0;
316315
ui64 dropBlobs = 0;
317316
auto& indexInfo = VersionedIndex.GetLastSchema()->GetIndexInfo();
318317
Y_ABORT_UNLESS(context.Changes->Tiering.emplace(pathId, ttl).second);
@@ -340,9 +339,8 @@ TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering
340339
continue;
341340
}
342341

343-
context.AllowEviction = (evictionSize <= context.MaxEvictBytes);
344342
context.AllowDrop = (dropBlobs <= TCompactionLimits::MAX_BLOBS_TO_DELETE);
345-
const bool tryEvictPortion = context.AllowEviction && ttl.HasTiers();
343+
const bool tryEvictPortion = ttl.HasTiers() && context.HasMemoryForEviction();
346344

347345
if (auto max = info->MaxValue(ttlColumnId)) {
348346
bool keep = !expireTimestampOpt;
@@ -390,8 +388,8 @@ TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering
390388
}
391389
if (currentTierName != tierName) {
392390
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "tiering switch detected")("from", currentTierName)("to", tierName);
393-
evictionSize += info->BlobsSizes().first;
394391
context.Changes->AddPortionToEvict(*info, TPortionEvictionFeatures(tierName, pathId, StoragesManager->GetOperator(tierName)));
392+
context.AppPortionForCheckMemoryUsage(*info);
395393
SignalCounters.OnPortionToEvict(info->BlobsBytes());
396394
}
397395
}
@@ -406,7 +404,7 @@ TDuration TColumnEngineForLogs::ProcessTiering(const ui64 pathId, const TTiering
406404
SignalCounters.OnPortionNoBorder(info->BlobsBytes());
407405
}
408406
}
409-
if (dWaiting > TDuration::MilliSeconds(500) && (!context.AllowEviction || !context.AllowDrop)) {
407+
if (dWaiting > TDuration::MilliSeconds(500) && (!context.HasMemoryForEviction() || !context.AllowDrop)) {
410408
dWaiting = TDuration::MilliSeconds(500);
411409
}
412410
Y_ABORT_UNLESS(!!dWaiting);
@@ -447,7 +445,7 @@ bool TColumnEngineForLogs::DrainEvictionQueue(std::map<TMonotonic, std::vector<T
447445
return hasChanges;
448446
}
449447

450-
std::shared_ptr<TTTLColumnEngineChanges> TColumnEngineForLogs::StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions, ui64 maxEvictBytes) noexcept {
448+
std::shared_ptr<TTTLColumnEngineChanges> TColumnEngineForLogs::StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions, const ui64 memoryUsageLimit) noexcept {
451449
AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD)("event", "StartTtl")("external", pathEviction.size())
452450
("internal", EvictionsController.MutableNextCheckInstantForTierings().size())
453451
;
@@ -456,7 +454,7 @@ std::shared_ptr<TTTLColumnEngineChanges> TColumnEngineForLogs::StartTtl(const TH
456454

457455
auto changes = std::make_shared<TTTLColumnEngineChanges>(TSplitSettings(), saverContext);
458456

459-
TTieringProcessContext context(maxEvictBytes, changes, busyPortions);
457+
TTieringProcessContext context(memoryUsageLimit, changes, busyPortions, TTTLColumnEngineChanges::BuildMemoryPredictor());
460458
bool hasExternalChanges = false;
461459
for (auto&& i : pathEviction) {
462460
context.DurationsForced[i.first] = ProcessTiering(i.first, i.second, context);
@@ -575,9 +573,11 @@ void TColumnEngineForLogs::DoRegisterTable(const ui64 pathId) {
575573
AFL_VERIFY(Tables.emplace(pathId, std::make_shared<TGranuleMeta>(pathId, GranulesStorage, SignalCounters.RegisterGranuleDataCounters(), VersionedIndex)).second);
576574
}
577575

578-
TColumnEngineForLogs::TTieringProcessContext::TTieringProcessContext(const ui64 maxEvictBytes, std::shared_ptr<TTTLColumnEngineChanges> changes, const THashSet<TPortionAddress>& busyPortions)
579-
: Now(TlsActivationContext ? AppData()->TimeProvider->Now() : TInstant::Now())
580-
, MaxEvictBytes(maxEvictBytes)
576+
TColumnEngineForLogs::TTieringProcessContext::TTieringProcessContext(const ui64 memoryUsageLimit,
577+
std::shared_ptr<TTTLColumnEngineChanges> changes, const THashSet<TPortionAddress>& busyPortions, const std::shared_ptr<TColumnEngineChanges::IMemoryPredictor>& memoryPredictor)
578+
: MemoryUsageLimit(memoryUsageLimit)
579+
, MemoryPredictor(memoryPredictor)
580+
, Now(TlsActivationContext ? AppData()->TimeProvider->Now() : TInstant::Now())
581581
, Changes(changes)
582582
, BusyPortions(busyPortions)
583583
{

‎ydb/core/tx/columnshard/engines/column_engine_logs.h

+22-7
Original file line numberDiff line numberDiff line change
@@ -84,20 +84,37 @@ class TColumnEngineForLogs : public IColumnEngine {
8484
std::shared_ptr<IStoragesManager> StoragesManager;
8585
TEvictionsController EvictionsController;
8686
class TTieringProcessContext {
87+
private:
88+
const ui64 MemoryUsageLimit;
89+
ui64 MemoryUsage = 0;
90+
std::shared_ptr<TColumnEngineChanges::IMemoryPredictor> MemoryPredictor;
8791
public:
8892
bool AllowEviction = true;
8993
bool AllowDrop = true;
9094
const TInstant Now;
91-
const ui64 MaxEvictBytes;
9295
std::shared_ptr<TTTLColumnEngineChanges> Changes;
9396
std::map<ui64, TDuration> DurationsForced;
9497
const THashSet<TPortionAddress>& BusyPortions;
95-
TTieringProcessContext(const ui64 maxEvictBytes, std::shared_ptr<TTTLColumnEngineChanges> changes, const THashSet<TPortionAddress>& busyPortions);
98+
99+
void AppPortionForCheckMemoryUsage(const TPortionInfo& info) {
100+
MemoryUsage = MemoryPredictor->AddPortion(info);
101+
}
102+
103+
bool HasMemoryForEviction() const {
104+
return MemoryUsage < MemoryUsageLimit;
105+
}
106+
107+
TTieringProcessContext(const ui64 memoryUsageLimit, std::shared_ptr<TTTLColumnEngineChanges> changes,
108+
const THashSet<TPortionAddress>& busyPortions, const std::shared_ptr<TColumnEngineChanges::IMemoryPredictor>& memoryPredictor);
96109
};
97110

98111
TDuration ProcessTiering(const ui64 pathId, const TTiering& tiering, TTieringProcessContext& context) const;
99112
bool DrainEvictionQueue(std::map<TMonotonic, std::vector<TEvictionsController::TTieringWithPathId>>& evictionsQueue, TTieringProcessContext& context) const;
100113
public:
114+
ui64* GetLastPortionPointer() {
115+
return &LastPortion;
116+
}
117+
101118
enum ETableIdx {
102119
GRANULES = 0,
103120
};
@@ -135,8 +152,8 @@ class TColumnEngineForLogs : public IColumnEngine {
135152
std::shared_ptr<TInsertColumnEngineChanges> StartInsert(std::vector<TInsertedData>&& dataToIndex) noexcept override;
136153
std::shared_ptr<TColumnEngineChanges> StartCompaction(const TCompactionLimits& limits, const THashSet<TPortionAddress>& busyPortions) noexcept override;
137154
std::shared_ptr<TCleanupColumnEngineChanges> StartCleanup(const TSnapshot& snapshot, THashSet<ui64>& pathsToDrop, ui32 maxRecords) noexcept override;
138-
std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction, const THashSet<TPortionAddress>& busyPortions,
139-
ui64 maxEvictBytes = TCompactionLimits::DEFAULT_EVICTION_BYTES) noexcept override;
155+
std::shared_ptr<TTTLColumnEngineChanges> StartTtl(const THashMap<ui64, TTiering>& pathEviction,
156+
const THashSet<TPortionAddress>& busyPortions, const ui64 memoryUsageLimit) noexcept override;
140157

141158
bool ApplyChanges(IDbWrapper& db, std::shared_ptr<TColumnEngineChanges> indexChanges,
142159
const TSnapshot& snapshot) noexcept override;
@@ -164,9 +181,7 @@ class TColumnEngineForLogs : public IColumnEngine {
164181
}
165182

166183
const TGranuleMeta& GetGranuleVerified(const ui64 pathId) const {
167-
auto it = Tables.find(pathId);
168-
AFL_VERIFY(it != Tables.end())("path_id", pathId)("count", Tables.size());
169-
return *it->second;
184+
return *GetGranulePtrVerified(pathId);
170185
}
171186

172187
std::shared_ptr<TGranuleMeta> GetGranulePtrVerified(const ui64 pathId) const {

‎ydb/core/tx/columnshard/engines/portions/portion_info.h

+3
Original file line numberDiff line numberDiff line change
@@ -364,6 +364,9 @@ class TPortionInfo {
364364
for (auto&& i : Records) {
365365
result.emplace(i.BlobRange.BlobId);
366366
}
367+
for (auto&& i : Indexes) {
368+
result.emplace(i.GetBlobRange().BlobId);
369+
}
367370
return result;
368371
}
369372

‎ydb/core/tx/columnshard/engines/reader/plain_reader/context.cpp

+4-1
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,10 @@ std::shared_ptr<NKikimr::NOlap::NPlainReader::IFetchingStep> TSpecialReadContext
7979
}
8080
current = current->AttachNext(std::make_shared<TFilterProgramStep>(i));
8181
}
82-
const TColumnsSet columnsAdditionalFetch = *FFColumns - *EFColumns - *SpecColumns - *PredicateColumns;
82+
TColumnsSet columnsAdditionalFetch = *FFColumns - *EFColumns - *SpecColumns;
83+
if (partialUsageByPredicate) {
84+
columnsAdditionalFetch = columnsAdditionalFetch - *PredicateColumns;
85+
}
8386
if (columnsAdditionalFetch.GetColumnsCount()) {
8487
current = current->AttachNext(std::make_shared<TBlobsFetchingStep>(std::make_shared<TColumnsSet>(columnsAdditionalFetch)));
8588
current = current->AttachNext(std::make_shared<TAssemblerStep>(std::make_shared<TColumnsSet>(columnsAdditionalFetch)));

0 commit comments

Comments
 (0)