Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Stop wide combiner state from growing unlimited #10997

Merged
merged 14 commits into from
Nov 1, 2024
Merged

Stop wide combiner state from growing unlimited #10997

Changes from 13 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
e5d9b3d
Stop state from growing and spill growed is spilling is enabled
vladl2802 Oct 28, 2024
2ef04c8
move growing into separate function and remove testing stuff
vladl2802 Oct 28, 2024
f576680
and unsafe grow into mkql_rh_hash and use it in wide combine
vladl2802 Oct 29, 2024
2fd06de
catch any exception in grow, remove debugging, move log in hashtable
vladl2802 Oct 29, 2024
0c2da0f
small refactor of grow
vladl2802 Oct 29, 2024
048c77b
small fix
vladl2802 Oct 29, 2024
927b46e
add AllowOutOfMemory in TState and use it for growing
vladl2802 Oct 29, 2024
f696a2b
rename new grow in hashtable
vladl2802 Oct 29, 2024
606659b
nano fix after renaming
vladl2802 Oct 29, 2024
2f0729a
move log from hashtable to tstate
vladl2802 Oct 29, 2024
54df553
remove useless include
vladl2802 Oct 29, 2024
0418f7d
revert to catching in state
vladl2802 Oct 30, 2024
4f0e9e7
catch exception always for log and rethrow it if we cant handle it
vladl2802 Oct 31, 2024
11a267a
make condition more readable
vladl2802 Oct 31, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 32 additions & 5 deletions ydb/library/yql/minikql/comp_nodes/mkql_wide_combine.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,8 +239,8 @@ class TState : public TComputationValue<TState> {
return KeyWidth + StateWidth;
}
public:
TState(TMemoryUsageInfo* memInfo, ui32 keyWidth, ui32 stateWidth, const THashFunc& hash, const TEqualsFunc& equal)
: TBase(memInfo), KeyWidth(keyWidth), StateWidth(stateWidth), States(hash, equal, CountRowsOnPage) {
TState(TMemoryUsageInfo* memInfo, ui32 keyWidth, ui32 stateWidth, const THashFunc& hash, const TEqualsFunc& equal, bool allowOutOfMemory = false)
: TBase(memInfo), KeyWidth(keyWidth), StateWidth(stateWidth), AllowOutOfMemory(allowOutOfMemory), States(hash, equal, CountRowsOnPage) {
CurrentPage = &Storage.emplace_back(RowSize() * CountRowsOnPage, NUdf::TUnboxedValuePod());
CurrentPosition = 0;
Tongue = CurrentPage->data();
Expand Down Expand Up @@ -275,11 +275,28 @@ class TState : public TComputationValue<TState> {
}
Throat = States.GetKey(itInsert) + KeyWidth;
if (isNew) {
States.CheckGrow();
GrowStates();
}
return isNew;
}

void GrowStates() {
try {
States.CheckGrow();
} catch (TMemoryLimitExceededException) {
YQL_LOG(INFO) << "State failed to grow";
if (IsOutOfMemory < AllowOutOfMemory) {
IsOutOfMemory = true;
} else {
throw;
}
}
}

bool CheckIsOutOfMemory() const {
return IsOutOfMemory;
}

template<bool SkipYields>
bool ReadMore() {
if constexpr (SkipYields) {
Expand Down Expand Up @@ -331,6 +348,8 @@ class TState : public TComputationValue<TState> {
private:
std::optional<TStorageIterator> ExtractIt;
const ui32 KeyWidth, StateWidth;
const bool AllowOutOfMemory;
bool IsOutOfMemory = false;
ui64 CurrentPosition = 0;
TRow* CurrentPage = nullptr;
TStorage Storage;
Expand Down Expand Up @@ -386,7 +405,7 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
const THashFunc& hash, const TEqualsFunc& equal, bool allowSpilling, TComputationContext& ctx
)
: TBase(memInfo)
, InMemoryProcessingState(memInfo, keyWidth, keyAndStateType->GetElementsCount() - keyWidth, hash, equal)
, InMemoryProcessingState(memInfo, keyWidth, keyAndStateType->GetElementsCount() - keyWidth, hash, equal, allowSpilling && ctx.SpillerFactory)
, UsedInputItemType(usedInputItemType)
, KeyAndStateType(keyAndStateType)
, KeyWidth(keyWidth)
Expand Down Expand Up @@ -451,6 +470,9 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
ETasteResult TasteIt() {
if (GetMode() == EOperatingMode::InMemory) {
bool isNew = InMemoryProcessingState.TasteIt();
if (InMemoryProcessingState.CheckIsOutOfMemory()) {
StateWantsToSpill = true;
}
Throat = InMemoryProcessingState.Throat;
return isNew ? ETasteResult::Init : ETasteResult::Update;
}
Expand Down Expand Up @@ -649,7 +671,11 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
}

bool CheckMemoryAndSwitchToSpilling() {
if (AllowSpilling && Ctx.SpillerFactory && IsSwitchToSpillingModeCondition()) {
if (!(AllowSpilling && Ctx.SpillerFactory)) {
return false;
}
if (StateWantsToSpill || IsSwitchToSpillingModeCondition()) {
StateWantsToSpill = false;
LogMemoryUsage();

SwitchMode(EOperatingMode::SplittingState);
Expand Down Expand Up @@ -840,6 +866,7 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
NUdf::TUnboxedValuePod* Throat = nullptr;

private:
bool StateWantsToSpill = false;
bool IsEverythingExtracted = false;

TState InMemoryProcessingState;
Expand Down
Loading