Skip to content

Commit d4ebd86

Browse files
ivanmorozov333ivanmorozov333
authored
categorized indexes (#15424)
1 parent 301cd78 commit d4ebd86

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

52 files changed

+1389
-117
lines changed

‎ydb/core/tx/columnshard/data_accessor/local_db/collector.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ void TCollector::DoAskData(
1111
}
1212
}
1313

14-
TDataCategorized TCollector::DoAnalyzeData(const std::vector<TPortionInfo::TConstPtr>& portions, const TString& /* consumer */) {
14+
TDataCategorized TCollector::DoAnalyzeData(const std::vector<TPortionInfo::TConstPtr>& portions, const TString& /*consumer*/) {
1515
TDataCategorized result;
1616
for (auto&& p : portions) {
1717
auto it = AccessorsCache.Find(p->GetPortionId());

‎ydb/core/tx/columnshard/engines/portions/data_accessor.cpp

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -372,6 +372,17 @@ ui64 TPortionDataAccessor::GetIndexRawBytes(const bool validation /*= true*/) co
372372
return sum;
373373
}
374374

375+
std::vector<const TIndexChunk*> TPortionDataAccessor::GetIndexChunksPointers(const ui32 indexId) const {
376+
std::vector<const TIndexChunk*> result;
377+
for (auto&& c : GetIndexesVerified()) {
378+
if (c.GetIndexId() == indexId) {
379+
AFL_VERIFY(c.GetChunkIdx() == result.size());
380+
result.emplace_back(&c);
381+
}
382+
}
383+
return result;
384+
}
385+
375386
std::vector<const TColumnRecord*> TPortionDataAccessor::GetColumnChunksPointers(const ui32 columnId) const {
376387
std::vector<const TColumnRecord*> result;
377388
for (auto&& c : GetRecordsVerified()) {

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,7 @@ class TPortionDataAccessor {
239239
NBlobOperations::NRead::TCompositeReadBlobs& blobs, const TIndexInfo& indexInfo) const;
240240

241241
std::vector<const TColumnRecord*> GetColumnChunksPointers(const ui32 columnId) const;
242+
std::vector<const TIndexChunk*> GetIndexChunksPointers(const ui32 indexId) const;
242243

243244
THashMap<TChunkAddress, TString> DecodeBlobAddresses(NBlobOperations::NRead::TCompositeReadBlobs&& blobs, const TIndexInfo& indexInfo) const;
244245

‎ydb/core/tx/columnshard/engines/protos/index.proto

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
package NKikimrTxColumnShard;
2+
3+
message TIndexCategoriesDescription {
4+
message TCategory {
5+
repeated uint64 Hashes = 1;
6+
optional uint32 FilterSize = 2;
7+
}
8+
repeated TCategory Categories = 1;
9+
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ PROTO_LIBRARY()
22

33
SRCS(
44
portion_info.proto
5+
index.proto
56
)
67

78
PEERDIR(

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/constructor.h

Lines changed: 37 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -2,44 +2,69 @@
22
#include "fetching.h"
33
#include "source.h"
44

5+
#include <ydb/core/formats/arrow/program/collection.h>
56
#include <ydb/core/tx/columnshard/blob.h>
67
#include <ydb/core/tx/columnshard/blobs_reader/task.h>
78
#include <ydb/core/tx/columnshard/engines/portions/column_record.h>
89
#include <ydb/core/tx/columnshard/engines/reader/abstract/read_context.h>
910
#include <ydb/core/tx/columnshard/engines/reader/abstract/read_metadata.h>
11+
#include <ydb/core/tx/columnshard/engines/scheme/indexes/abstract/collection.h>
1012

1113
namespace NKikimr::NOlap::NReader::NCommon {
1214

15+
class TFetchingResultContext {
16+
private:
17+
NArrow::NAccessor::TAccessorsCollection& Accessors;
18+
NIndexes::TIndexesCollection& Indexes;
19+
std::shared_ptr<IDataSource> Source;
20+
21+
public:
22+
NArrow::NAccessor::TAccessorsCollection& GetAccessors() {
23+
return Accessors;
24+
}
25+
NIndexes::TIndexesCollection& GetIndexes() const {
26+
return Indexes;
27+
}
28+
const std::shared_ptr<IDataSource>& GetSource() const {
29+
return Source;
30+
}
31+
TFetchingResultContext(
32+
NArrow::NAccessor::TAccessorsCollection& accessors, NIndexes::TIndexesCollection& indexes, const std::shared_ptr<IDataSource>& source)
33+
: Accessors(accessors)
34+
, Indexes(indexes)
35+
, Source(source)
36+
{
37+
}
38+
};
39+
1340
class IKernelFetchLogic {
1441
private:
1542
YDB_READONLY(ui32, ColumnId, 0);
1643

17-
virtual void DoStart(TReadActionsCollection& nextRead) = 0;
44+
virtual void DoStart(TReadActionsCollection& nextRead, TFetchingResultContext& context) = 0;
1845
virtual void DoOnDataReceived(TReadActionsCollection& nextRead, NBlobOperations::NRead::TCompositeReadBlobs& blobs) = 0;
19-
virtual void DoOnDataCollected() = 0;
46+
virtual void DoOnDataCollected(TFetchingResultContext& context) = 0;
2047

2148
protected:
22-
const std::shared_ptr<IDataSource> Source;
23-
const std::shared_ptr<NArrow::NAccessor::TAccessorsCollection> Resources;
49+
const std::shared_ptr<IStoragesManager> StoragesManager;
2450

2551
public:
2652
virtual ~IKernelFetchLogic() = default;
2753

28-
IKernelFetchLogic(const ui32 columnId, const std::shared_ptr<IDataSource>& source)
54+
IKernelFetchLogic(const ui32 columnId, const std::shared_ptr<IStoragesManager>& storagesManager)
2955
: ColumnId(columnId)
30-
, Source(source)
31-
, Resources(Source->GetStageData().GetTable()) {
32-
AFL_VERIFY(Resources);
56+
, StoragesManager(storagesManager) {
57+
AFL_VERIFY(StoragesManager);
3358
}
3459

35-
void Start(TReadActionsCollection& nextRead) {
36-
DoStart(nextRead);
60+
void Start(TReadActionsCollection& nextRead, TFetchingResultContext& context) {
61+
DoStart(nextRead, context);
3762
}
3863
void OnDataReceived(TReadActionsCollection& nextRead, NBlobOperations::NRead::TCompositeReadBlobs& blobs) {
3964
DoOnDataReceived(nextRead, blobs);
4065
}
41-
void OnDataCollected() {
42-
DoOnDataCollected();
66+
void OnDataCollected(TFetchingResultContext& context) {
67+
DoOnDataCollected(context);
4368
}
4469
};
4570

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/default_fetching.h

Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -42,15 +42,16 @@ class TDefaultFetchLogic: public IKernelFetchLogic {
4242

4343
std::vector<TChunkRestoreInfo> ColumnChunks;
4444
std::optional<TString> StorageId;
45-
virtual void DoOnDataCollected() override {
45+
virtual void DoOnDataCollected(TFetchingResultContext& context) override {
4646
AFL_VERIFY(!IIndexInfo::IsSpecialColumn(GetColumnId()));
4747
std::vector<TPortionDataAccessor::TAssembleBlobInfo> chunks;
4848
for (auto&& i : ColumnChunks) {
4949
chunks.emplace_back(i.ExtractDataVerified());
5050
}
5151

52-
TPortionDataAccessor::TPreparedColumn column(std::move(chunks), Source->GetSourceSchema()->GetColumnLoaderVerified(GetColumnId()));
53-
Resources->AddVerified(GetColumnId(), column.AssembleAccessor().DetachResult(), true);
52+
TPortionDataAccessor::TPreparedColumn column(
53+
std::move(chunks), context.GetSource()->GetSourceSchema()->GetColumnLoaderVerified(GetColumnId()));
54+
context.GetAccessors().AddVerified(GetColumnId(), column.AssembleAccessor().DetachResult(), true);
5455
}
5556

5657
virtual void DoOnDataReceived(TReadActionsCollection& /*nextRead*/, NBlobOperations::NRead::TCompositeReadBlobs& blobs) override {
@@ -66,42 +67,43 @@ class TDefaultFetchLogic: public IKernelFetchLogic {
6667
}
6768
}
6869

69-
virtual void DoStart(TReadActionsCollection& nextRead) override {
70-
auto columnChunks = Source->GetStageData().GetPortionAccessor().GetColumnChunksPointers(GetColumnId());
70+
virtual void DoStart(TReadActionsCollection& nextRead, TFetchingResultContext& context) override {
71+
auto source = context.GetSource();
72+
auto columnChunks = source->GetStageData().GetPortionAccessor().GetColumnChunksPointers(GetColumnId());
7173
if (columnChunks.empty()) {
72-
ColumnChunks.emplace_back(Source->GetRecordsCount(), TPortionDataAccessor::TAssembleBlobInfo(Source->GetRecordsCount(),
73-
Source->GetSourceSchema()->GetExternalDefaultValueVerified(GetColumnId())));
74+
ColumnChunks.emplace_back(source->GetRecordsCount(), TPortionDataAccessor::TAssembleBlobInfo(source->GetRecordsCount(),
75+
source->GetSourceSchema()->GetExternalDefaultValueVerified(GetColumnId())));
7476
return;
7577
}
76-
StorageId = Source->GetColumnStorageId(GetColumnId());
77-
TBlobsAction blobsAction(Source->GetContext()->GetCommonContext()->GetStoragesManager(), NBlobOperations::EConsumer::SCAN);
78+
StorageId = source->GetColumnStorageId(GetColumnId());
79+
TBlobsAction blobsAction(source->GetContext()->GetCommonContext()->GetStoragesManager(), NBlobOperations::EConsumer::SCAN);
7880
auto reading = blobsAction.GetReading(*StorageId);
79-
auto filterPtr = Source->GetStageData().GetAppliedFilter();
81+
auto filterPtr = source->GetStageData().GetAppliedFilter();
8082
const NArrow::TColumnFilter& cFilter = filterPtr ? *filterPtr : NArrow::TColumnFilter::BuildAllowFilter();
81-
auto itFilter = cFilter.GetIterator(false, Source->GetRecordsCount());
83+
auto itFilter = cFilter.GetIterator(false, source->GetRecordsCount());
8284
bool itFinished = false;
8385
for (auto&& c : columnChunks) {
8486
AFL_VERIFY(!itFinished);
8587
if (!itFilter.IsBatchForSkip(c->GetMeta().GetRecordsCount())) {
8688
reading->SetIsBackgroundProcess(false);
87-
reading->AddRange(Source->RestoreBlobRange(c->BlobRange));
88-
ColumnChunks.emplace_back(c->GetMeta().GetRecordsCount(), Source->RestoreBlobRange(c->BlobRange));
89+
reading->AddRange(source->RestoreBlobRange(c->BlobRange));
90+
ColumnChunks.emplace_back(c->GetMeta().GetRecordsCount(), source->RestoreBlobRange(c->BlobRange));
8991
} else {
9092
ColumnChunks.emplace_back(
9193
c->GetMeta().GetRecordsCount(), TPortionDataAccessor::TAssembleBlobInfo(c->GetMeta().GetRecordsCount(),
92-
Source->GetSourceSchema()->GetExternalDefaultValueVerified(c->GetColumnId())));
94+
source->GetSourceSchema()->GetExternalDefaultValueVerified(c->GetColumnId())));
9395
}
9496
itFinished = !itFilter.Next(c->GetMeta().GetRecordsCount());
9597
}
96-
AFL_VERIFY(itFinished)("filter", itFilter.DebugString())("count", Source->GetRecordsCount());
98+
AFL_VERIFY(itFinished)("filter", itFilter.DebugString())("count", source->GetRecordsCount());
9799
for (auto&& i : blobsAction.GetReadingActions()) {
98100
nextRead.Add(i);
99101
}
100102
}
101103

102104
public:
103-
TDefaultFetchLogic(const ui32 columnId, const std::shared_ptr<IDataSource>& source)
104-
: TBase(columnId, source) {
105+
TDefaultFetchLogic(const ui32 columnId, const std::shared_ptr<IStoragesManager>& storagesManager)
106+
: TBase(columnId, storagesManager) {
105107
}
106108
};
107109

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/fetched_data.h

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
#include <ydb/core/tx/columnshard/blobs_reader/task.h>
1010
#include <ydb/core/tx/columnshard/engines/portions/data_accessor.h>
1111
#include <ydb/core/tx/columnshard/engines/portions/portion_info.h>
12+
#include <ydb/core/tx/columnshard/engines/scheme/indexes/abstract/collection.h>
1213
#include <ydb/core/tx/limiter/grouped_memory/usage/abstract.h>
1314

1415
#include <ydb/library/accessor/accessor.h>
@@ -28,6 +29,7 @@ class TFetchedData {
2829
TFetchers Fetchers;
2930
YDB_ACCESSOR_DEF(TBlobs, Blobs);
3031
YDB_READONLY_DEF(std::shared_ptr<NArrow::NAccessor::TAccessorsCollection>, Table);
32+
YDB_READONLY_DEF(std::shared_ptr<NIndexes::TIndexesCollection>, Indexes);
3133
YDB_READONLY(bool, Aborted, false);
3234

3335
std::shared_ptr<NGroupedMemoryManager::TAllocationGuard> AccessorsGuard;
@@ -69,6 +71,7 @@ class TFetchedData {
6971
TFetchedData(const bool useFilter, const ui32 recordsCount) {
7072
Table = std::make_shared<NArrow::NAccessor::TAccessorsCollection>(recordsCount);
7173
Table->SetFilterUsage(useFilter);
74+
Indexes = std::make_shared<NIndexes::TIndexesCollection>();
7275
}
7376

7477
void SetAccessorsGuard(std::shared_ptr<NGroupedMemoryManager::TAllocationGuard>&& guard) {
@@ -118,7 +121,8 @@ class TFetchedData {
118121
return result;
119122
}
120123

121-
void AddBatch(const std::shared_ptr<NArrow::TGeneralContainer>& container, const NArrow::NSSA::IColumnResolver& resolver, const bool withFilter) {
124+
void AddBatch(
125+
const std::shared_ptr<NArrow::TGeneralContainer>& container, const NArrow::NSSA::IColumnResolver& resolver, const bool withFilter) {
122126
Table->AddBatch(container, resolver, withFilter);
123127
}
124128

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/fetching.cpp

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -194,6 +194,7 @@ void TFetchingScriptBuilder::AddAssembleStep(
194194
TConclusion<bool> TProgramStepPrepare::DoExecuteInplace(const std::shared_ptr<IDataSource>& source, const TFetchingScriptCursor& step) const {
195195
TReadActionsCollection readActions;
196196
THashMap<ui32, std::shared_ptr<IKernelFetchLogic>> fetchers;
197+
TFetchingResultContext context(*source->GetStageData().GetTable(), *source->GetStageData().GetIndexes(), source);
197198
for (auto&& i : Step.GetOriginalColumnsToUse()) {
198199
const auto columnLoader = source->GetSourceSchema()->GetColumnLoaderVerified(i.GetColumnId());
199200
auto customFetchInfo = Step->BuildFetchTask(i.GetColumnId(), columnLoader->GetAccessorConstructor()->GetType(), source->GetStageData().GetTable());
@@ -205,12 +206,12 @@ TConclusion<bool> TProgramStepPrepare::DoExecuteInplace(const std::shared_ptr<ID
205206
}
206207
std::shared_ptr<IKernelFetchLogic> logic;
207208
if (customFetchInfo->GetFullRestore() || source->GetStageData().GetPortionAccessor().GetColumnChunksPointers(i.GetColumnId()).empty()) {
208-
logic = std::make_shared<TDefaultFetchLogic>(i.GetColumnId(), source);
209+
logic = std::make_shared<TDefaultFetchLogic>(i.GetColumnId(), source->GetContext()->GetCommonContext()->GetStoragesManager());
209210
} else {
210211
AFL_VERIFY(customFetchInfo->GetSubColumns().size());
211212
logic = std::make_shared<TSubColumnsFetchLogic>(i.GetColumnId(), source, customFetchInfo->GetSubColumns());
212213
}
213-
logic->Start(readActions);
214+
logic->Start(readActions, context);
214215
AFL_VERIFY(fetchers.emplace(i.GetColumnId(), logic).second)("column_id", i.GetColumnId());
215216
}
216217
if (readActions.IsEmpty()) {
@@ -240,6 +241,7 @@ TConclusion<bool> TProgramStep::DoExecuteInplace(const std::shared_ptr<IDataSour
240241

241242
TConclusion<bool> TProgramStepAssemble::DoExecuteInplace(
242243
const std::shared_ptr<IDataSource>& source, const TFetchingScriptCursor& /*cursor*/) const {
244+
TFetchingResultContext context(*source->GetStageData().GetTable(), *source->GetStageData().GetIndexes(), source);
243245
for (auto&& i : Step.GetOriginalColumnsToUse()) {
244246
const auto columnLoader = source->GetSourceSchema()->GetColumnLoaderVerified(i.GetColumnId());
245247
auto customFetchInfo =
@@ -248,7 +250,7 @@ TConclusion<bool> TProgramStepAssemble::DoExecuteInplace(
248250
continue;
249251
}
250252
auto fetcher = source->MutableStageData().ExtractFetcherVerified(i.GetColumnId());
251-
fetcher->OnDataCollected();
253+
fetcher->OnDataCollected(context);
252254
}
253255
return true;
254256
}

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/sub_columns_fetching.h

Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -169,18 +169,19 @@ class TSubColumnsFetchLogic: public IKernelFetchLogic {
169169
std::vector<TColumnChunkRestoreInfo> ColumnChunks;
170170
std::optional<TString> StorageId;
171171
bool NeedToAddResource = false;
172-
virtual void DoOnDataCollected() override {
172+
virtual void DoOnDataCollected(TFetchingResultContext& context) override {
173173
if (NeedToAddResource) {
174174
NArrow::NAccessor::TCompositeChunkedArray::TBuilder compositeBuilder(ChunkExternalInfo.GetColumnType());
175175
for (auto&& i : ColumnChunks) {
176-
i.Finish(nullptr, Source);
176+
i.Finish(nullptr, context.GetSource());
177177
compositeBuilder.AddChunk(i.GetPartialArray());
178178
}
179-
Resources->AddVerified(GetColumnId(), compositeBuilder.Finish(), true);
179+
context.GetAccessors().AddVerified(GetColumnId(), compositeBuilder.Finish(), true);
180180
} else {
181181
ui32 pos = 0;
182182
for (auto&& i : ColumnChunks) {
183-
i.Finish(std::make_shared<NArrow::TColumnFilter>(Resources->GetAppliedFilter()->Slice(pos, i.GetRecordsCount())), Source);
183+
i.Finish(std::make_shared<NArrow::TColumnFilter>(context.GetAccessors().GetAppliedFilter()->Slice(pos, i.GetRecordsCount())),
184+
context.GetSource());
184185
pos += i.GetRecordsCount();
185186
}
186187
}
@@ -189,7 +190,7 @@ class TSubColumnsFetchLogic: public IKernelFetchLogic {
189190
virtual void DoOnDataReceived(TReadActionsCollection& nextRead, NBlobOperations::NRead::TCompositeReadBlobs& blobs) override {
190191
AFL_VERIFY(ColumnChunks.size());
191192
AFL_VERIFY(!!StorageId);
192-
TBlobsAction blobsAction(Source->GetContext()->GetCommonContext()->GetStoragesManager(), NBlobOperations::EConsumer::SCAN);
193+
TBlobsAction blobsAction(StoragesManager, NBlobOperations::EConsumer::SCAN);
193194
auto reading = blobsAction.GetReading(*StorageId);
194195
reading->SetIsBackgroundProcess(false);
195196
for (auto&& i : ColumnChunks) {
@@ -228,19 +229,20 @@ class TSubColumnsFetchLogic: public IKernelFetchLogic {
228229
nextRead.Add(reading);
229230
}
230231

231-
virtual void DoStart(TReadActionsCollection& nextRead) override {
232-
auto columnChunks = Source->GetStageData().GetPortionAccessor().GetColumnChunksPointers(GetColumnId());
232+
virtual void DoStart(TReadActionsCollection& nextRead, TFetchingResultContext& context) override {
233+
auto source = context.GetSource();
234+
auto columnChunks = source->GetStageData().GetPortionAccessor().GetColumnChunksPointers(GetColumnId());
233235
AFL_VERIFY(columnChunks.size());
234-
StorageId = Source->GetColumnStorageId(GetColumnId());
235-
TBlobsAction blobsAction(Source->GetContext()->GetCommonContext()->GetStoragesManager(), NBlobOperations::EConsumer::SCAN);
236+
StorageId = source->GetColumnStorageId(GetColumnId());
237+
TBlobsAction blobsAction(StoragesManager, NBlobOperations::EConsumer::SCAN);
236238
auto reading = blobsAction.GetReading(*StorageId);
237239
reading->SetIsBackgroundProcess(false);
238-
auto filterPtr = Source->GetStageData().GetAppliedFilter();
240+
auto filterPtr = source->GetStageData().GetAppliedFilter();
239241
const NArrow::TColumnFilter& cFilter = filterPtr ? *filterPtr : NArrow::TColumnFilter::BuildAllowFilter();
240-
auto itFilter = cFilter.GetIterator(false, Source->GetRecordsCount());
242+
auto itFilter = cFilter.GetIterator(false, source->GetRecordsCount());
241243
bool itFinished = false;
242244

243-
auto accessor = Resources->GetAccessorOptional(GetColumnId());
245+
auto accessor = context.GetAccessors().GetAccessorOptional(GetColumnId());
244246
NeedToAddResource = !accessor;
245247
std::vector<std::shared_ptr<NArrow::NAccessor::IChunkedArray>> chunks;
246248
if (!NeedToAddResource) {
@@ -256,7 +258,7 @@ class TSubColumnsFetchLogic: public IKernelFetchLogic {
256258
auto& meta = columnChunks[chunkIdx]->GetMeta();
257259
AFL_VERIFY(!itFinished);
258260
if (!itFilter.IsBatchForSkip(meta.GetRecordsCount())) {
259-
const TBlobRange range = Source->RestoreBlobRange(columnChunks[chunkIdx]->BlobRange);
261+
const TBlobRange range = source->RestoreBlobRange(columnChunks[chunkIdx]->BlobRange);
260262
ColumnChunks.emplace_back(range, ChunkExternalInfo.GetSubset(meta.GetRecordsCount()));
261263
if (!NeedToAddResource) {
262264
AFL_VERIFY(resChunkIdx < chunks.size())("chunks", chunks.size())("meta", columnChunks.size())("need", NeedToAddResource);
@@ -270,18 +272,18 @@ class TSubColumnsFetchLogic: public IKernelFetchLogic {
270272
itFinished = !itFilter.Next(meta.GetRecordsCount());
271273
}
272274
AFL_VERIFY(NeedToAddResource || (resChunkIdx == chunks.size()));
273-
AFL_VERIFY(itFinished)("filter", itFilter.DebugString())("count", Source->GetRecordsCount());
275+
AFL_VERIFY(itFinished)("filter", itFilter.DebugString())("count", source->GetRecordsCount());
274276
for (auto&& i : blobsAction.GetReadingActions()) {
275277
nextRead.Add(i);
276278
}
277279
}
278280

279281
public:
280282
TSubColumnsFetchLogic(const ui32 columnId, const std::shared_ptr<IDataSource>& source, const std::vector<TString>& subColumns)
281-
: TBase(columnId, source)
282-
, ChunkExternalInfo(Source->GetSourceSchema()->GetColumnLoaderVerified(GetColumnId())->BuildAccessorContext(Source->GetRecordsCount()))
283+
: TBase(columnId, source->GetContext()->GetCommonContext()->GetStoragesManager())
284+
, ChunkExternalInfo(source->GetSourceSchema()->GetColumnLoaderVerified(GetColumnId())->BuildAccessorContext(source->GetRecordsCount()))
283285
, SubColumns(subColumns) {
284-
const auto loader = Source->GetSourceSchema()->GetColumnLoaderVerified(GetColumnId());
286+
const auto loader = source->GetSourceSchema()->GetColumnLoaderVerified(GetColumnId());
285287
AFL_VERIFY(loader->GetAccessorConstructor()->GetType() == NArrow::NAccessor::IChunkedArray::EType::SubColumnsArray)
286288
("type", loader->GetAccessorConstructor()->GetType());
287289
}

0 commit comments

Comments
 (0)