Skip to content

Commit 0ae7561

Browse files
swalrus1swalrus1
authored
Merge 58946f7 into d1adfd1 
2 parents d1adfd1 + 58946f7 commit 0ae7561

File tree

14 files changed

+189
-180
lines changed

14 files changed

+189
-180
lines changed

‎ydb/core/tx/columnshard/counters/portion_index.cpp

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ void TPortionIndexStats::RemovePortion(const NOlap::TPortionInfo& portion) {
2525

2626
{
2727
auto findClass = TotalStats.find(portionClass);
28-
AFL_VERIFY(!findClass.IsEnd())("path_id", portion.GetPathId());
28+
AFL_VERIFY(findClass != TotalStats.end())("path_id", portion.GetPathId());
2929
findClass->second.RemovePortion(portion);
3030
if (findClass->second.IsEmpty()) {
3131
TotalStats.erase(findClass);
@@ -34,9 +34,9 @@ void TPortionIndexStats::RemovePortion(const NOlap::TPortionInfo& portion) {
3434

3535
{
3636
auto findPathId = StatsByPathId.find(portion.GetPathId());
37-
AFL_VERIFY(!findPathId.IsEnd())("path_id", portion.GetPathId());
37+
AFL_VERIFY(findPathId != StatsByPathId.end())("path_id", portion.GetPathId());
3838
auto findClass = findPathId->second.find(portionClass);
39-
AFL_VERIFY(!findClass.IsEnd())("path_id", portion.GetPathId());
39+
AFL_VERIFY(findClass != findPathId->second.end())("path_id", portion.GetPathId());
4040
findClass->second.RemovePortion(portion);
4141
if (findClass->second.IsEmpty()) {
4242
findPathId->second.erase(findClass);

‎ydb/core/tx/columnshard/engines/reader/common/description.h

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@ struct TReadDescription {
3131
std::vector<ui32> ColumnIds;
3232

3333
const std::shared_ptr<IScanCursor>& GetScanCursor() const {
34-
AFL_VERIFY(ScanCursor);
3534
return ScanCursor;
3635
}
3736

‎ydb/core/tx/columnshard/engines/reader/common_reader/constructor/read_metadata.cpp

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,13 @@ TConclusionStatus TReadMetadata::Init(
4242
return TConclusionStatus::Success();
4343
}
4444

45+
TReadMetadata::TReadMetadata(const std::shared_ptr<TVersionedIndex>& schemaIndex, const TReadDescription& read)
46+
: TBase(schemaIndex, read.PKRangesFilter->IsReverse() ? TReadMetadataBase::ESorting::DESC : TReadMetadataBase::ESorting::ASC,
47+
read.GetProgram(), schemaIndex->GetSchemaVerified(read.GetSnapshot()), read.GetSnapshot(), read.GetScanCursor())
48+
, PathId(read.PathId)
49+
, ReadStats(std::make_shared<TReadStats>()) {
50+
}
51+
4552
std::set<ui32> TReadMetadata::GetEarlyFilterColumnIds() const {
4653
auto& indexInfo = ResultIndexSchema->GetIndexInfo();
4754
const auto& ids = GetProgram().GetEarlyFilterColumns();

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

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -117,12 +117,7 @@ class TReadMetadata: public TReadMetadataBase {
117117
NYql::NDqProto::EDqStatsMode StatsMode = NYql::NDqProto::EDqStatsMode::DQ_STATS_MODE_NONE;
118118
std::shared_ptr<TReadStats> ReadStats;
119119

120-
TReadMetadata(const ui64 pathId, const std::shared_ptr<TVersionedIndex> info, const TSnapshot& snapshot, const ESorting sorting,
121-
const TProgramContainer& ssaProgram, const std::shared_ptr<IScanCursor>& scanCursor)
122-
: TBase(info, sorting, ssaProgram, info->GetSchemaVerified(snapshot), snapshot, scanCursor)
123-
, PathId(pathId)
124-
, ReadStats(std::make_shared<TReadStats>()) {
125-
}
120+
TReadMetadata(const std::shared_ptr<TVersionedIndex>& schemaIndex, const TReadDescription& read);
126121

127122
virtual std::vector<TNameTypeInfo> GetKeyYqlSchema() const override {
128123
return GetResultSchema()->GetIndexInfo().GetPrimaryKeyColumns();

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,15 @@ class TColumnsSetIds {
8585
}
8686
return result;
8787
}
88+
89+
TColumnsSetIds& operator+=(const TColumnsSetIds& external) {
90+
return (*this = *this + external);
91+
}
92+
93+
TColumnsSetIds& operator-=(const TColumnsSetIds& external) {
94+
return (*this = *this - external);
95+
}
96+
8897
bool IsEmpty() const {
8998
return ColumnIds.empty();
9099
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,7 @@ class TOptionalAssemblerStep: public IFetchingStep {
126126
class TColumnBlobsFetchingStep: public IFetchingStep {
127127
private:
128128
using TBase = IFetchingStep;
129-
TColumnsSetIds Columns;
129+
YDB_READONLY_DEF(TColumnsSetIds, Columns);
130130

131131
protected:
132132
virtual TConclusion<bool> DoExecuteInplace(const std::shared_ptr<IDataSource>& source, const TFetchingScriptCursor& step) const override;

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

Lines changed: 32 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -87,12 +87,9 @@ TString TFetchingScript::DebugString() const {
8787
return sb;
8888
}
8989

90-
TFetchingScript::TFetchingScript(const TSpecialReadContext& /*context*/) {
91-
}
92-
93-
void TFetchingScript::Allocation(const std::set<ui32>& entityIds, const EStageFeaturesIndexes stage, const EMemType mType) {
90+
void TFetchingScriptBuilder::AddAllocation(const std::set<ui32>& entityIds, const EStageFeaturesIndexes stage, const EMemType mType) {
9491
if (Steps.size() == 0) {
95-
AddStep<TAllocateMemoryStep>(entityIds, mType, stage);
92+
AddStep(std::make_shared<TAllocateMemoryStep>(entityIds, mType, stage));
9693
} else {
9794
std::optional<ui32> addIndex;
9895
for (i32 i = Steps.size() - 1; i >= 0; --i) {
@@ -125,42 +122,52 @@ TString IFetchingStep::DebugString() const {
125122
return sb;
126123
}
127124

128-
bool TColumnsAccumulator::AddFetchingStep(TFetchingScript& script, const TColumnsSetIds& columns, const EStageFeaturesIndexes stage) {
129-
auto actualColumns = GetNotFetchedAlready(columns);
130-
FetchingReadyColumns = FetchingReadyColumns + (TColumnsSetIds)columns;
131-
if (!actualColumns.IsEmpty()) {
132-
script.Allocation(columns.GetColumnIds(), stage, EMemType::Blob);
133-
script.AddStep(std::make_shared<TColumnBlobsFetchingStep>(actualColumns));
134-
return true;
125+
TFetchingScriptBuilder::TFetchingScriptBuilder(const TSpecialReadContext& context)
126+
: GuaranteeNotOptional(context.GetMergeColumns())
127+
, FullSchema(context.GetReadMetadata()->GetResultSchema()) {
128+
}
129+
130+
void TFetchingScriptBuilder::AddFetchingStep(const TColumnsSetIds& columns, const EStageFeaturesIndexes stage) {
131+
auto actualColumns = columns - AddedFetchingColumns;
132+
AddedFetchingColumns += columns;
133+
if (actualColumns.IsEmpty()) {
134+
return;
135+
}
136+
if (Steps.size() && std::dynamic_pointer_cast<TColumnBlobsFetchingStep>(Steps.back())) {
137+
TColumnsSetIds fetchingColumns = actualColumns + std::dynamic_pointer_cast<TColumnBlobsFetchingStep>(Steps.back())->GetColumns();
138+
Steps.pop_back();
139+
AddAllocation(actualColumns.GetColumnIds(), stage, EMemType::Blob);
140+
AddStep(std::make_shared<TColumnBlobsFetchingStep>(fetchingColumns));
141+
} else {
142+
AddAllocation(actualColumns.GetColumnIds(), stage, EMemType::Blob);
143+
AddStep(std::make_shared<TColumnBlobsFetchingStep>(actualColumns));
135144
}
136-
return false;
137145
}
138146

139-
bool TColumnsAccumulator::AddAssembleStep(
140-
TFetchingScript& script, const TColumnsSetIds& columns, const TString& purposeId, const EStageFeaturesIndexes stage, const bool sequential) {
141-
auto actualColumns = columns - AssemblerReadyColumns;
142-
AssemblerReadyColumns = AssemblerReadyColumns + columns;
147+
void TFetchingScriptBuilder::AddAssembleStep(
148+
const TColumnsSetIds& columns, const TString& purposeId, const EStageFeaturesIndexes stage, const bool sequential) {
149+
auto actualColumns = columns - AddedAssembleColumns;
150+
AddedAssembleColumns += columns;
143151
if (actualColumns.IsEmpty()) {
144-
return false;
152+
return;
145153
}
146154
auto actualSet = std::make_shared<TColumnsSet>(actualColumns.GetColumnIds(), FullSchema);
147155
if (sequential) {
148156
const auto notSequentialColumnIds = GuaranteeNotOptional->Intersect(*actualSet);
149157
if (notSequentialColumnIds.size()) {
150-
script.Allocation(notSequentialColumnIds, stage, EMemType::Raw);
158+
AddAllocation(notSequentialColumnIds, stage, EMemType::Raw);
151159
std::shared_ptr<TColumnsSet> cross = actualSet->BuildSamePtr(notSequentialColumnIds);
152-
script.AddStep<TAssemblerStep>(cross, purposeId);
160+
AddStep(std::make_shared<TAssemblerStep>(cross, purposeId));
153161
*actualSet = *actualSet - *cross;
154162
}
155163
if (!actualSet->IsEmpty()) {
156-
script.Allocation(notSequentialColumnIds, stage, EMemType::RawSequential);
157-
script.AddStep<TOptionalAssemblerStep>(actualSet, purposeId);
164+
AddAllocation(notSequentialColumnIds, stage, EMemType::RawSequential);
165+
AddStep(std::make_shared<TOptionalAssemblerStep>(actualSet, purposeId));
158166
}
159167
} else {
160-
script.Allocation(actualColumns.GetColumnIds(), stage, EMemType::Raw);
161-
script.AddStep<TAssemblerStep>(actualSet, purposeId);
168+
AddAllocation(actualColumns.GetColumnIds(), stage, EMemType::Raw);
169+
AddStep(std::make_shared<TAssemblerStep>(actualSet, purposeId));
162170
}
163-
return true;
164171
}
165172

166173
TConclusion<bool> TProgramStep::DoExecuteInplace(const std::shared_ptr<IDataSource>& source, const TFetchingScriptCursor& /*step*/) const {

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

Lines changed: 36 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,10 @@ class IFetchingStep: public TNonCopyable {
9595
Signals.AddBytes(size);
9696
}
9797

98+
virtual bool Merge(const std::shared_ptr<const IFetchingStep>& /*nextStep*/) {
99+
return false;
100+
}
101+
98102
virtual ~IFetchingStep() = default;
99103

100104
[[nodiscard]] TConclusion<bool> ExecuteInplace(const std::shared_ptr<IDataSource>& source, const TFetchingScriptCursor& step) const {
@@ -115,15 +119,16 @@ class IFetchingStep: public TNonCopyable {
115119

116120
class TFetchingScript {
117121
private:
118-
YDB_ACCESSOR(TString, BranchName, "UNDEFINED");
122+
YDB_READONLY_DEF(TString, BranchName);
119123
std::vector<std::shared_ptr<IFetchingStep>> Steps;
120124
TAtomic StartInstant;
121125
TAtomic FinishInstant;
122126

123127
public:
124-
TFetchingScript(const TSpecialReadContext& context);
125-
126-
void Allocation(const std::set<ui32>& entityIds, const EStageFeaturesIndexes stage, const EMemType mType);
128+
TFetchingScript(const TString& branchName, std::vector<std::shared_ptr<IFetchingStep>>&& steps)
129+
: BranchName(branchName)
130+
, Steps(std::move(steps)) {
131+
}
127132

128133
void AddStepDataSize(const ui32 index, const ui64 size) {
129134
GetStep(index)->AddDataSize(size);
@@ -145,26 +150,6 @@ class TFetchingScript {
145150
return Steps[index];
146151
}
147152

148-
template <class T, typename... Args>
149-
std::shared_ptr<T> AddStep(Args... args) {
150-
auto result = std::make_shared<T>(args...);
151-
Steps.emplace_back(result);
152-
return result;
153-
}
154-
155-
template <class T, typename... Args>
156-
std::shared_ptr<T> InsertStep(const ui32 index, Args... args) {
157-
AFL_VERIFY(index <= Steps.size())("index", index)("size", Steps.size());
158-
auto result = std::make_shared<T>(args...);
159-
Steps.insert(Steps.begin() + index, result);
160-
return result;
161-
}
162-
163-
void AddStep(const std::shared_ptr<IFetchingStep>& step) {
164-
AFL_VERIFY(step);
165-
Steps.emplace_back(step);
166-
}
167-
168153
bool IsFinished(const ui32 currentStepIdx) const {
169154
AFL_VERIFY(currentStepIdx <= Steps.size());
170155
return currentStepIdx == Steps.size();
@@ -231,26 +216,41 @@ class TFetchingScriptOwner: TNonCopyable {
231216
}
232217
};
233218

234-
class TColumnsAccumulator {
219+
class TFetchingScriptBuilder {
235220
private:
236-
TColumnsSetIds FetchingReadyColumns;
237-
TColumnsSetIds AssemblerReadyColumns;
238-
ISnapshotSchema::TPtr FullSchema;
239221
std::shared_ptr<TColumnsSetIds> GuaranteeNotOptional;
222+
ISnapshotSchema::TPtr FullSchema;
223+
224+
YDB_ACCESSOR(TString, BranchName, "UNDEFINED");
225+
std::vector<std::shared_ptr<IFetchingStep>> Steps;
226+
YDB_READONLY_DEF(TColumnsSetIds, AddedFetchingColumns);
227+
YDB_READONLY_DEF(TColumnsSetIds, AddedAssembleColumns);
228+
229+
private:
230+
void AddAllocation(const std::set<ui32>& entityIds, const EStageFeaturesIndexes stage, const EMemType mType);
231+
232+
template <class T, typename... Args>
233+
std::shared_ptr<T> InsertStep(const ui32 index, Args... args) {
234+
AFL_VERIFY(index <= Steps.size())("index", index)("size", Steps.size());
235+
auto result = std::make_shared<T>(args...);
236+
Steps.insert(Steps.begin() + index, result);
237+
return result;
238+
}
240239

241240
public:
242-
TColumnsAccumulator(const std::shared_ptr<TColumnsSetIds>& guaranteeNotOptional, const ISnapshotSchema::TPtr& fullSchema)
243-
: FullSchema(fullSchema)
244-
, GuaranteeNotOptional(guaranteeNotOptional) {
241+
TFetchingScriptBuilder(const TSpecialReadContext& context);
242+
243+
std::shared_ptr<TFetchingScript> Build()&& {
244+
return std::make_shared<TFetchingScript>(BranchName, std::move(Steps));
245245
}
246246

247-
TColumnsSetIds GetNotFetchedAlready(const TColumnsSetIds& columns) const {
248-
return columns - FetchingReadyColumns;
247+
void AddStep(const std::shared_ptr<IFetchingStep>& step) {
248+
AFL_VERIFY(step);
249+
Steps.emplace_back(step);
249250
}
250251

251-
bool AddFetchingStep(TFetchingScript& script, const TColumnsSetIds& columns, const EStageFeaturesIndexes stage);
252-
bool AddAssembleStep(TFetchingScript& script, const TColumnsSetIds& columns, const TString& purposeId, const EStageFeaturesIndexes stage,
253-
const bool sequential);
252+
void AddFetchingStep(const TColumnsSetIds& columns, const EStageFeaturesIndexes stage);
253+
void AddAssembleStep(const TColumnsSetIds& columns, const TString& purposeId, const EStageFeaturesIndexes stage, const bool sequential);
254254
};
255255

256256
class TFetchingScriptCursor {

‎ydb/core/tx/columnshard/engines/reader/common_reader/iterator/ya.make

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ SRCS(
1313

1414
PEERDIR(
1515
ydb/core/tx/columnshard/engines/scheme
16+
yql/essentials/minikql
1617
)
1718

1819
GENERATE_ENUM_SERIALIZATION(columns_set.h)

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

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,7 @@ NKikimr::TConclusion<std::shared_ptr<TReadMetadataBase>> TIndexScannerConstructo
3535

3636
TDataStorageAccessor dataAccessor(insertTable, index);
3737
AFL_VERIFY(read.PathId);
38-
auto readMetadata = std::make_shared<TReadMetadata>(read.PathId, index->CopyVersionedIndexPtr(), read.GetSnapshot(),
39-
IsReverse ? TReadMetadataBase::ESorting::DESC : TReadMetadataBase::ESorting::ASC, read.GetProgram(), nullptr);
38+
auto readMetadata = std::make_shared<TReadMetadata>(index->CopyVersionedIndexPtr(), read);
4039

4140
auto initResult = readMetadata->Init(self, read, dataAccessor);
4241
if (!initResult) {

0 commit comments

Comments
 (0)