Skip to content

Commit 0612e30

Browse files
lll-phill-llllll-phill-lll
authored
disabled block engine for spilling (#7778)
1 parent 9812ae2 commit 0612e30

File tree

18 files changed

+147
-40
lines changed

18 files changed

+147
-40
lines changed

‎ydb/core/kqp/opt/logical/kqp_opt_log.cpp

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,8 @@ class TKqpLogicalOptTransformer : public TOptimizeTransformerBase {
118118
true, // defaultWatermarksMode
119119
true); // syncActor
120120
} else {
121-
output = DqRewriteAggregate(node, ctx, TypesCtx, false, KqpCtx.Config->HasOptEnableOlapPushdown() || KqpCtx.Config->HasOptUseFinalizeByKey(), KqpCtx.Config->HasOptUseFinalizeByKey());
121+
NDq::TSpillingSettings spillingSettings(KqpCtx.Config->GetEnabledSpillingNodes());
122+
output = DqRewriteAggregate(node, ctx, TypesCtx, false, KqpCtx.Config->HasOptEnableOlapPushdown() || KqpCtx.Config->HasOptUseFinalizeByKey(), KqpCtx.Config->HasOptUseFinalizeByKey(), spillingSettings.IsAggregationSpillingEnabled());
122123
}
123124
if (output) {
124125
DumpAppliedRule("RewriteAggregate", node.Ptr(), output.Cast().Ptr(), ctx);

‎ydb/core/kqp/opt/physical/kqp_opt_phy.cpp

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -252,7 +252,8 @@ class TKqpPhysicalOptTransformer : public TOptimizeTransformerBase {
252252
}
253253

254254
TMaybeNode<TExprBase> ExpandAggregatePhase(TExprBase node, TExprContext& ctx) {
255-
auto output = ExpandAggregatePeepholeImpl(node.Ptr(), ctx, TypesCtx, KqpCtx.Config->HasOptUseFinalizeByKey(), false);
255+
NDq::TSpillingSettings spillingSettings(KqpCtx.Config->GetEnabledSpillingNodes());
256+
auto output = ExpandAggregatePeepholeImpl(node.Ptr(), ctx, TypesCtx, KqpCtx.Config->HasOptUseFinalizeByKey(), false, spillingSettings.IsAggregationSpillingEnabled());
256257
DumpAppliedRule("ExpandAggregatePhase", node.Ptr(), output, ctx);
257258
return TExprBase(output);
258259
}

‎ydb/core/kqp/provider/yql_kikimr_settings.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ ui64 ParseEnableSpillingNodes(const TString &v) {
3434
if (s.empty()) {
3535
throw yexception() << "Empty value item";
3636
}
37-
auto value = FromString<NYql::TDqSettings::EEnabledSpillingNodes>(s);
37+
auto value = FromString<NDq::EEnabledSpillingNodes>(s);
3838
res |= ui64(value);
3939
}
4040
return res;

‎ydb/core/kqp/ut/olap/aggregations_ut.cpp

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,49 @@ Y_UNIT_TEST_SUITE(KqpOlapAggregations) {
171171
}
172172
}
173173

174+
Y_UNIT_TEST_TWIN(DisableBlockEngineInAggregationWithSpilling, AllowSpilling) {
175+
auto settings = TKikimrSettings()
176+
.SetWithSampleTables(false);
177+
settings.AppConfig.MutableTableServiceConfig()->SetBlockChannelsMode(NKikimrConfig::TTableServiceConfig_EBlockChannelsMode_BLOCK_CHANNELS_FORCE);
178+
if (AllowSpilling) {
179+
settings.AppConfig.MutableTableServiceConfig()->SetEnableSpillingNodes("Aggregation");
180+
} else {
181+
settings.AppConfig.MutableTableServiceConfig()->SetEnableSpillingNodes("None");
182+
}
183+
TKikimrRunner kikimr(settings);
184+
185+
TLocalHelper(kikimr).CreateTestOlapTable();
186+
auto client = kikimr.GetQueryClient();
187+
188+
{
189+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 10000, 3000000, 1000);
190+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 11000, 3001000, 1000);
191+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 12000, 3002000, 1000);
192+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 13000, 3003000, 1000);
193+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 14000, 3004000, 1000);
194+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 20000, 2000000, 7000);
195+
WriteTestData(kikimr, "/Root/olapStore/olapTable", 30000, 1000000, 11000);
196+
}
197+
198+
{
199+
TString query = R"(
200+
--!syntax_v1
201+
SELECT
202+
COUNT(*)
203+
FROM `/Root/olapStore/olapTable`
204+
GROUP BY level
205+
)";
206+
207+
auto res = StreamExplainQuery(query, client);
208+
UNIT_ASSERT_C(res.IsSuccess(), res.GetIssues().ToString());
209+
210+
auto plan = CollectStreamResult(res);
211+
212+
bool hasWideCombiner = plan.QueryStats->Getquery_ast().Contains("WideCombiner");
213+
UNIT_ASSERT_C(hasWideCombiner == AllowSpilling, plan.QueryStats->Getquery_ast());
214+
}
215+
}
216+
174217
Y_UNIT_TEST_TWIN(CountAllPushdown, UseLlvm) {
175218
auto settings = TKikimrSettings()
176219
.SetWithSampleTables(false);

‎ydb/core/kqp/ut/olap/kqp_olap_ut.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2461,6 +2461,7 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
24612461
NKikimrConfig::TAppConfig appConfig;
24622462
appConfig.MutableTableServiceConfig()->SetEnableOlapSink(true);
24632463
appConfig.MutableTableServiceConfig()->SetBlockChannelsMode(blockChannelsMode);
2464+
appConfig.MutableTableServiceConfig()->SetEnableSpillingNodes("None");
24642465
auto settings = TKikimrSettings()
24652466
.SetAppConfig(appConfig)
24662467
.SetWithSampleTables(true);

‎ydb/library/yql/core/yql_aggregate_expander.cpp

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ TExprNode::TPtr TAggregateExpander::ExpandAggregateWithFullOutput()
3636

3737
HaveDistinct = AnyOf(AggregatedColumns->ChildrenList(),
3838
[](const auto& child) { return child->ChildrenSize() == 3; });
39-
EffectiveCompact = (HaveDistinct && CompactForDistinct && !TypesCtx.IsBlockEngineEnabled()) || ForceCompact || HasSetting(*settings, "compact");
39+
EffectiveCompact = (HaveDistinct && CompactForDistinct && !UseBlocks) || ForceCompact || HasSetting(*settings, "compact");
4040
for (const auto& trait : Traits) {
4141
auto mergeLambda = trait->Child(5);
4242
if (mergeLambda->Tail().IsCallable("Void")) {
@@ -67,7 +67,7 @@ TExprNode::TPtr TAggregateExpander::ExpandAggregateWithFullOutput()
6767
return GeneratePhases();
6868
}
6969

70-
if (TypesCtx.IsBlockEngineEnabled()) {
70+
if (UseBlocks) {
7171
if (Suffix == "Combine") {
7272
auto ret = TryGenerateBlockCombine();
7373
if (ret) {
@@ -2785,7 +2785,7 @@ TExprNode::TPtr TAggregateExpander::GeneratePhases() {
27852785
streams.push_back(SerializeIdxSet(indicies));
27862786
}
27872787

2788-
if (TypesCtx.IsBlockEngineEnabled()) {
2788+
if (UseBlocks) {
27892789
for (ui32 i = 0; i < unionAllInputs.size(); ++i) {
27902790
unionAllInputs[i] = Ctx.Builder(Node->Pos())
27912791
.Callable("Map")
@@ -2806,7 +2806,7 @@ TExprNode::TPtr TAggregateExpander::GeneratePhases() {
28062806
}
28072807

28082808
auto settings = cleanOutputSettings;
2809-
if (TypesCtx.IsBlockEngineEnabled()) {
2809+
if (UseBlocks) {
28102810
settings = AddSetting(*settings, Node->Pos(), "many_streams", Ctx.NewList(Node->Pos(), std::move(streams)), Ctx);
28112811
}
28122812

@@ -2839,7 +2839,7 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockCombine() {
28392839
}
28402840

28412841
TExprNode::TPtr TAggregateExpander::TryGenerateBlockMergeFinalize() {
2842-
if (UsePartitionsByKeys || !TypesCtx.IsBlockEngineEnabled()) {
2842+
if (UsePartitionsByKeys || !UseBlocks) {
28432843
return nullptr;
28442844
}
28452845

@@ -2934,7 +2934,7 @@ TExprNode::TPtr ExpandAggregatePeephole(const TExprNode::TPtr& node, TExprContex
29342934
return ret;
29352935
}
29362936
}
2937-
return ExpandAggregatePeepholeImpl(node, ctx, typesCtx, false, typesCtx.IsBlockEngineEnabled());
2937+
return ExpandAggregatePeepholeImpl(node, ctx, typesCtx, false, typesCtx.IsBlockEngineEnabled(), false);
29382938
}
29392939

29402940
} // namespace NYql

‎ydb/library/yql/core/yql_aggregate_expander.h

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ namespace NYql {
99
class TAggregateExpander {
1010
public:
1111
TAggregateExpander(bool usePartitionsByKeys, const bool useFinalizeByKeys, const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx,
12-
bool forceCompact = false, bool compactForDistinct = false, bool usePhases = false)
12+
bool forceCompact = false, bool compactForDistinct = false, bool usePhases = false, bool allowSpilling = false)
1313
: Node(node)
1414
, Ctx(ctx)
1515
, TypesCtx(typesCtx)
@@ -25,6 +25,7 @@ class TAggregateExpander {
2525
, HaveSessionSetting(false)
2626
, OriginalRowType(nullptr)
2727
, RowType(nullptr)
28+
, UseBlocks(typesCtx.IsBlockEngineEnabled() && !allowSpilling)
2829
{
2930
PreMap = Ctx.Builder(node->Pos())
3031
.Lambda()
@@ -115,6 +116,7 @@ class TAggregateExpander {
115116
const TStructExprType* RowType;
116117
TVector<const TItemExprType*> RowItems;
117118
TExprNode::TPtr PreMap;
119+
bool UseBlocks;
118120

119121
TExprNode::TListType InitialColumnNames;
120122
TExprNode::TListType FinalColumnNames;
@@ -130,8 +132,10 @@ class TAggregateExpander {
130132
std::unordered_map<std::string_view, TExprNode::TPtr> UdfWasChanged;
131133
};
132134

133-
inline TExprNode::TPtr ExpandAggregatePeepholeImpl(const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx, const bool useFinalizeByKey, const bool useBlocks) {
134-
TAggregateExpander aggExpander(!useFinalizeByKey && !useBlocks, useFinalizeByKey, node, ctx, typesCtx, true);
135+
inline TExprNode::TPtr ExpandAggregatePeepholeImpl(const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx,
136+
const bool useFinalizeByKey, const bool useBlocks, const bool allowSpilling) {
137+
TAggregateExpander aggExpander(!useFinalizeByKey && !useBlocks, useFinalizeByKey, node, ctx, typesCtx,
138+
true, false, false, allowSpilling);
135139
return aggExpander.ExpandAggregate();
136140
}
137141

‎ydb/library/yql/dq/common/dq_common.h

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,34 @@ enum class EHashJoinMode {
9292
GraceAndSelf /* "graceandself" */,
9393
};
9494

95+
enum class EEnabledSpillingNodes : ui64 {
96+
None = 0ULL /* "None" */,
97+
GraceJoin = 1ULL /* "GraceJoin" */,
98+
Aggregation = 2ULL /* "Aggregation" */,
99+
All = ~0ULL /* "All" */,
100+
};
101+
102+
class TSpillingSettings {
103+
public:
104+
TSpillingSettings() = default;
105+
explicit TSpillingSettings(ui64 mask) : Mask(mask) {};
106+
107+
operator bool() const {
108+
return Mask;
109+
}
110+
111+
bool IsGraceJoinSpillingEnabled() const {
112+
return Mask & ui64(EEnabledSpillingNodes::GraceJoin);
113+
}
114+
115+
bool IsAggregationSpillingEnabled() const {
116+
return Mask & ui64(EEnabledSpillingNodes::Aggregation);
117+
}
118+
119+
private:
120+
const ui64 Mask = 0;
121+
};
122+
95123
} // namespace NYql::NDq
96124

97125
IOutputStream& operator<<(IOutputStream& stream, const NYql::NDq::TTxId& txId);

‎ydb/library/yql/dq/opt/dq_opt_log.cpp

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,13 @@ using namespace NYql::NNodes;
1717
namespace NYql::NDq {
1818

1919
TExprBase DqRewriteAggregate(TExprBase node, TExprContext& ctx, TTypeAnnotationContext& typesCtx, bool compactForDistinct,
20-
bool usePhases, const bool useFinalizeByKey)
20+
bool usePhases, const bool useFinalizeByKey, const bool allowSpilling)
2121
{
2222
if (!node.Maybe<TCoAggregateBase>()) {
2323
return node;
2424
}
25-
TAggregateExpander aggExpander(!typesCtx.IsBlockEngineEnabled() && !useFinalizeByKey, useFinalizeByKey, node.Ptr(), ctx, typesCtx, false, compactForDistinct, usePhases);
25+
TAggregateExpander aggExpander(!typesCtx.IsBlockEngineEnabled() && !useFinalizeByKey,
26+
useFinalizeByKey, node.Ptr(), ctx, typesCtx, false, compactForDistinct, usePhases, allowSpilling);
2627
auto result = aggExpander.ExpandAggregate();
2728
YQL_ENSURE(result);
2829

‎ydb/library/yql/dq/opt/dq_opt_log.h

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,8 @@ namespace NYql {
1919

2020
namespace NYql::NDq {
2121

22-
NNodes::TExprBase DqRewriteAggregate(NNodes::TExprBase node, TExprContext& ctx, TTypeAnnotationContext& typesCtx, bool compactForDistinct, bool usePhases, const bool useFinalizeByKey);
22+
NNodes::TExprBase DqRewriteAggregate(NNodes::TExprBase node, TExprContext& ctx, TTypeAnnotationContext& typesCtx,
23+
bool compactForDistinct, bool usePhases, const bool useFinalizeByKey, const bool allowSpilling);
2324

2425
NNodes::TExprBase DqRewriteTakeSortToTopSort(NNodes::TExprBase node, TExprContext& ctx, const TParentsMap& parents);
2526

0 commit comments

Comments
 (0)