add tests to pragma and feature flag

Author: gridnevvvit

ydb/core/kqp/compute_actor/kqp_compute_actor_factory.cpp

@@ -166,6 +166,10 @@ class TKqpCaFactory : public IKqpNodeComputeActorFactory {
         runtimeSettings.UseSpilling = args.WithSpilling;
         runtimeSettings.StatsMode = args.StatsMode;
 
+        if (runtimeSettings.UseSpilling) {
+            args.Task->SetEnableSpilling(runtimeSettings.UseSpilling);
+        }
+
         if (args.Deadline) {
             runtimeSettings.Timeout = args.Deadline - TAppData::TimeProvider->Now();
         }

ydb/core/kqp/executer_actor/kqp_planner.cpp

@@ -194,6 +194,7 @@ std::unique_ptr<TEvKqpNode::TEvStartKqpTasksRequest> TKqpPlanner::SerializeReque
     request.SetStartAllOrFail(true);
     if (UseDataQueryPool) {
         request.MutableRuntimeSettings()->SetExecType(NYql::NDqProto::TComputeRuntimeSettings::DATA);
+        request.MutableRuntimeSettings()->SetUseSpilling(WithSpilling);
     } else {
         request.MutableRuntimeSettings()->SetExecType(NYql::NDqProto::TComputeRuntimeSettings::SCAN);
         request.MutableRuntimeSettings()->SetUseSpilling(WithSpilling);

ydb/core/kqp/ut/spilling/kqp_scan_spilling_ut.cpp

@@ -32,10 +32,70 @@ NKikimrConfig::TAppConfig AppCfg() {
     return appCfg;
 }
 
+NKikimrConfig::TAppConfig AppCfgLowComputeLimits(ui64 reasonableTreshold) {
+    NKikimrConfig::TAppConfig appCfg;
+
+    auto* rm = appCfg.MutableTableServiceConfig()->MutableResourceManager();
+    rm->SetMkqlLightProgramMemoryLimit(100);
+    rm->SetMkqlHeavyProgramMemoryLimit(300);
+    rm->SetReasonableSpillingTreshold(reasonableTreshold);
+    appCfg.MutableTableServiceConfig()->SetEnableQueryServiceSpilling(true);
+
+    auto* spilling = appCfg.MutableTableServiceConfig()->MutableSpillingServiceConfig()->MutableLocalFileConfig();
+
+    spilling->SetEnable(true);
+    spilling->SetRoot("./spilling/");
+
+    return appCfg;
+}
+
+
 } // anonymous namespace
 
 Y_UNIT_TEST_SUITE(KqpScanSpilling) {
 
+Y_UNIT_TEST_TWIN(SpillingInRuntimeNodes, EnabledSpilling) {
+    ui64 reasonableTreshold = EnabledSpilling ? 100 : 200_MB;
+    Cerr << "cwd: " << NFs::CurrentWorkingDirectory() << Endl;
+    TKikimrRunner kikimr(AppCfgLowComputeLimits(reasonableTreshold));
+
+    auto db = kikimr.GetQueryClient();
+
+    for (ui32 i = 0; i < 300; ++i) {
+        auto result = db.ExecuteQuery(Sprintf(R"(
+            --!syntax_v1
+            REPLACE INTO `/Root/KeyValue` (Key, Value) VALUES (%d, "%s")
+        )", i, TString(200000 + i, 'a' + (i % 26)).c_str()), NYdb::NQuery::TTxControl::BeginTx().CommitTx()).GetValueSync();
+        UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+    }
+
+    auto query = R"(
+        --!syntax_v1
+        PRAGMA ydb.EnableSpillingNodes="GraceJoin";
+        select t1.Key, t1.Value, t2.Key, t2.Value
+        from `/Root/KeyValue` as t1 full join `/Root/KeyValue` as t2 on t1.Value = t2.Value
+        order by t1.Value
+    )";
+
+    auto explainMode = NYdb::NQuery::TExecuteQuerySettings().ExecMode(NYdb::NQuery::EExecMode::Explain);
+    auto planres = db.ExecuteQuery(query, NYdb::NQuery::TTxControl::NoTx(), explainMode).ExtractValueSync();
+    UNIT_ASSERT_VALUES_EQUAL_C(planres.GetStatus(), EStatus::SUCCESS, planres.GetIssues().ToString());
+
+    Cerr << planres.GetStats()->GetAst() << Endl;
+
+    auto result = db.ExecuteQuery(query, NYdb::NQuery::TTxControl::BeginTx().CommitTx(), NYdb::NQuery::TExecuteQuerySettings()).ExtractValueSync();
+    UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+    TKqpCounters counters(kikimr.GetTestServer().GetRuntime()->GetAppData().Counters);
+    if (EnabledSpilling) {
+        UNIT_ASSERT(counters.SpillingWriteBlobs->Val() > 0);
+        UNIT_ASSERT(counters.SpillingReadBlobs->Val() > 0);
+    } else {
+        UNIT_ASSERT(counters.SpillingWriteBlobs->Val() == 0);
+        UNIT_ASSERT(counters.SpillingReadBlobs->Val() == 0);
+    }
+}
+
 Y_UNIT_TEST(SelfJoinQueryService) {
     Cerr << "cwd: " << NFs::CurrentWorkingDirectory() << Endl;
 

ydb/library/yql/dq/actors/compute/dq_sync_compute_actor_base.h

@@ -55,7 +55,7 @@ class TDqSyncComputeActorBase: public TDqComputeActorBase<TDerived, TComputeActo
         }
     }
 
-    bool DoHandleChannelsAfterFinishImpl() override final{ 
+    bool DoHandleChannelsAfterFinishImpl() override final{
         Y_ABORT_UNLESS(this->Checkpoints);
 
         if (this->Checkpoints->HasPendingCheckpoint() && !this->Checkpoints->ComputeActorStateSaved() && ReadyToCheckpoint()) {