Intermediate changes

.github/workflows/nightly_run.yaml

@@ -40,6 +40,5 @@ jobs:
     secrets: inherit
     with:
       build_target: ydb/
-      sanitizer: none
       run_build: true
       run_tests: true

.mapping.json

@@ -8800,6 +8800,11 @@
   "ydb/library/yql/utils/simd/exec/stream_store/CMakeLists.linux-x86_64.txt":"",
   "ydb/library/yql/utils/simd/exec/stream_store/CMakeLists.txt":"",
   "ydb/library/yql/utils/simd/exec/stream_store/CMakeLists.windows-x86_64.txt":"",
+  "ydb/library/yql/utils/simd/exec/tuples_to_bucket/CMakeLists.darwin-x86_64.txt":"",
+  "ydb/library/yql/utils/simd/exec/tuples_to_bucket/CMakeLists.linux-aarch64.txt":"",
+  "ydb/library/yql/utils/simd/exec/tuples_to_bucket/CMakeLists.linux-x86_64.txt":"",
+  "ydb/library/yql/utils/simd/exec/tuples_to_bucket/CMakeLists.txt":"",
+  "ydb/library/yql/utils/simd/exec/tuples_to_bucket/CMakeLists.windows-x86_64.txt":"",
   "ydb/library/yql/utils/simd/ut/CMakeLists.darwin-x86_64.txt":"",
   "ydb/library/yql/utils/simd/ut/CMakeLists.linux-aarch64.txt":"",
   "ydb/library/yql/utils/simd/ut/CMakeLists.linux-x86_64.txt":"",

build/conf/project_specific/yql_udf.conf

@@ -36,12 +36,16 @@ macro UDF_NO_PROBE() {
 module YQL_UDF_TEST: PY3TEST_BIN {
     SET_APPEND(_MAKEFILE_INCLUDE_LIKE_DEPS canondata/result.json)
 
-    PEERDIR(ydb/library/yql/tests/common/udf_test)
+    PEERDIR(contrib/ydb/library/yql/tests/common/udf_test)
 
-    DEPENDS(ydb/library/yql/tools/astdiff)
-    DEPENDS(ydb/library/yql/tools/yqlrun)
-    DATA(arcadia/ydb/library/yql/mount)
-    DATA(arcadia/ydb/library/yql/cfg/udf_test)
+    DEPENDS(contrib/ydb/library/yql/tools/astdiff)
+    DEPENDS(contrib/ydb/library/yql/tools/yqlrun)
+    DATA(arcadia/contrib/ydb/library/yql/mount)
+    DATA(arcadia/contrib/ydb/library/yql/cfg/udf_test)
+    ENV(YQL_ASTDIFF_PATH="contrib/ydb/library/yql/tools/astdiff/astdiff")
+    ENV(YQL_CONFIG_DIR="contrib/ydb/library/yql/cfg/udf_test")
+    ENV(YQL_YQLRUN_PATH="contrib/ydb/library/yql/tools/yqlrun/yqlrun")
+    ENV(YQL_SQL2YQL_PATH="contrib/ydb/library/yql/tools/sql2yql/sql2yql")
 }
 
 module YQL_UDF_YDB_TEST: PY3TEST_BIN {
@@ -74,8 +78,8 @@ module YQL_UDF_TEST_CONTRIB: PY3TEST_BIN {
 ###
 ### https://yql.yandex-team.ru/docs/yt/udf/cpp/
 macro _ADD_YQL_UDF_DEPS() {
-    PEERDIR(ydb/library/yql/public/udf)
-    PEERDIR(ydb/library/yql/public/udf/support)
+    PEERDIR(contrib/ydb/library/yql/public/udf)
+    PEERDIR(contrib/ydb/library/yql/public/udf/support)
 }
 
 macro _ADD_YQL_UDF_YDB_DEPS() {
@@ -97,7 +101,7 @@ macro _MAKE_YQL_UDF() {
     _ADD_YQL_UDF_DEPS()
     SET_APPEND(USER_CXXFLAGS -DBUILD_UDF)
     # For Windows using declspecs
-    DEFAULT(YQL_UDF_EXPORT ${ARCADIA_ROOT}/ydb/library/yql/public/udf/udfs_exports.exports)
+    DEFAULT(YQL_UDF_EXPORT ${ARCADIA_ROOT}/contrib/ydb/library/yql/public/udf/udfs_exports.exports)
 
     when ($WINDOWS == "yes") {
         YQL_UDF_EXPORT=

build/plugins/yql_python_udf.py

@@ -19,7 +19,7 @@ def onregister_yql_python_udf(unit, *args):
 
     unit.onyql_abi_version(['2', '27', '0'])
     unit.onpeerdir(['yql/udfs/common/python/python_udf'])
-    unit.onpeerdir(['ydb/library/yql/public/udf'])
+    unit.onpeerdir(['contrib/ydb/library/yql/public/udf'])
 
     if add_libra_modules:
         unit.onpeerdir(['quality/user_sessions/libra_arc/noyql'])
@@ -37,7 +37,7 @@ def onregister_yql_python_udf(unit, *args):
 
     output_includes = [
         'yql/udfs/common/python/python_udf/python_udf.h',
-        'ydb/library/yql/public/udf/udf_registrator.h',
+        'contrib/ydb/library/yql/public/udf/udf_registrator.h',
     ]
     if add_libra_modules:
         output_includes.append('yql/udfs/quality/libra/module/module.h')

build/scripts/gen_yql_python_udf.py

@@ -3,7 +3,7 @@ import sys
 TEMPLATE="""
 #include <yql/udfs/common/python/python_udf/python_udf.h>
 
-#include <ydb/library/yql/public/udf/udf_registrator.h>
+#include <contrib/ydb/library/yql/public/udf/udf_registrator.h>
 
 #if @WITH_LIBRA@
 #include <yql/udfs/quality/libra/module/module.h>

ydb/core/kqp/opt/kqp_query_plan.cpp

@@ -1975,9 +1975,8 @@ TString AddExecStatsToTxPlan(const TString& txPlanJson, const NYql::NDqProto::TD
 
         SetNonZero(node, "ComputeTimeUs", taskStats.GetComputeCpuTimeUs());
 
-        SetNonZero(node, "WaitTimeUs", taskStats.GetWaitTimeUs());                   // need to be reviewed
-        SetNonZero(node, "PendingInputTimeUs", taskStats.GetPendingInputTimeUs());   // need to be reviewed
-        SetNonZero(node, "PendingOutputTimeUs", taskStats.GetPendingOutputTimeUs()); // need to be reviewed
+        SetNonZero(node, "WaitInputTimeUs", taskStats.GetWaitInputTimeUs());
+        SetNonZero(node, "WaitOutputTimeUs", taskStats.GetWaitOutputTimeUs());
 
         NKqpProto::TKqpTaskExtraStats taskExtraStats;
         if (taskStats.GetExtra().UnpackTo(&taskExtraStats)) {

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

@@ -625,8 +625,10 @@ protected:
 
     void InternalError(NYql::NDqProto::StatusIds::StatusCode statusCode, TIssues issues) {
         CA_LOG_E(InternalErrorLogString(statusCode, issues));
-        TaskRunner->GetAllocatorPtr()->InvalidateMemInfo();
-        TaskRunner->GetAllocatorPtr()->DisableStrictAllocationCheck();
+        if (TaskRunner) {
+            TaskRunner->GetAllocatorPtr()->InvalidateMemInfo();
+            TaskRunner->GetAllocatorPtr()->DisableStrictAllocationCheck();
+        }
         std::optional<TGuard<NKikimr::NMiniKQL::TScopedAlloc>> guard = MaybeBindAllocator();
         State = NDqProto::COMPUTE_STATE_FAILURE;
         ReportStateAndMaybeDie(statusCode, issues);
@@ -1940,11 +1942,7 @@ private:
     }
 
     virtual const NYql::NDq::TTaskRunnerStatsBase* GetTaskRunnerStats() {
-        if (!TaskRunner) {
-            return nullptr;
-        }
-        TaskRunner->UpdateStats();
-        return TaskRunner->GetStats();
+        return TaskRunner ? TaskRunner->GetStats() : nullptr;
     }
 
     virtual const IDqAsyncOutputBuffer* GetSink(ui64, const TAsyncOutputInfoBase& sinkInfo) const {

ydb/library/yql/dq/actors/compute/dq_compute_actor_stats.cpp

@@ -44,14 +44,8 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TTaskRunnerStatsBase&
     protoTask->SetComputeCpuTimeUs(taskStats.ComputeCpuTime.MicroSeconds());
     protoTask->SetBuildCpuTimeUs(taskStats.BuildCpuTime.MicroSeconds());
 
-    protoTask->SetWaitTimeUs(taskStats.WaitTime.MicroSeconds());             // to be reviewed
-    protoTask->SetWaitOutputTimeUs(taskStats.WaitOutputTime.MicroSeconds()); // to be reviewed
-
-    // All run statuses metrics
-    protoTask->SetPendingInputTimeUs(taskStats.RunStatusTimeMetrics[ERunStatus::PendingInput].MicroSeconds());   // to be reviewed
-    protoTask->SetPendingOutputTimeUs(taskStats.RunStatusTimeMetrics[ERunStatus::PendingOutput].MicroSeconds()); // to be reviewed
-    protoTask->SetFinishTimeUs(taskStats.RunStatusTimeMetrics[ERunStatus::Finished].MicroSeconds());             // to be reviewed
-    static_assert(TRunStatusTimeMetrics::StatusesCount == 3); // Add all statuses here
+    protoTask->SetWaitInputTimeUs(taskStats.WaitInputTime.MicroSeconds());
+    protoTask->SetWaitOutputTimeUs(taskStats.WaitOutputTime.MicroSeconds());
 
     if (StatsLevelCollectProfile(level)) {
         if (taskStats.ComputeCpuTimeByRun) {
@@ -106,6 +100,7 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TTaskRunnerStatsBase&
                     }
                     {
                         auto& protoChannel = *protoTask->AddInputChannels();
+                        protoChannel.SetChannelId(pushStats.ChannelId); // only one of ids
                         protoChannel.SetSrcStageId(srcStageId);
                         FillAsyncStats(*protoChannel.MutablePush(), pushStats);
                         FillAsyncStats(*protoChannel.MutablePop(), popStats);
@@ -135,8 +130,8 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TTaskRunnerStatsBase&
     //
     // task runner is not aware of ingress/egress stats, fill in in CA
     //
-    for (auto& [inputIndex, sources] : taskStats.Sources) {
-        if (StatsLevelCollectFull(level)) {
+    if (StatsLevelCollectFull(level)) {
+        for (auto& [inputIndex, sources] : taskStats.Sources) {
             auto& protoSource = *protoTask->AddSources();
             protoSource.SetInputIndex(inputIndex);
             FillAsyncStats(*protoSource.MutablePush(), sources->GetPushStats());
@@ -184,6 +179,7 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TTaskRunnerStatsBase&
                     }
                     {
                         auto& protoChannel = *protoTask->AddOutputChannels();
+                        protoChannel.SetChannelId(popStats.ChannelId); // only one of ids
                         protoChannel.SetDstStageId(dstStageId);
                         FillAsyncStats(*protoChannel.MutablePush(), pushStats);
                         FillAsyncStats(*protoChannel.MutablePop(), popStats);

ydb/library/yql/dq/actors/protos/dq_stats.proto

@@ -205,11 +205,12 @@ message TDqTaskStats {
     uint64 StartTimeMs = 158;
     uint64 FinishTimeMs = 5;     // task finish time, timestamp in millis
     uint64 FirstRowTimeMs = 4;   // first row time, timestamp in millis
-    uint64 WaitTimeUs = 104;       // total wait (input + output) wall time
+    reserved 104;
+    uint64 WaitInputTimeUs = 111;  // wait input wall time  (any input: channels, source, ...)
     uint64 WaitOutputTimeUs = 105; // wait output wall time (any output: channels, sinks, ...)
-    uint64 PendingInputTimeUs = 107; // time waiting input data
-    uint64 PendingOutputTimeUs = 108; // time waiting output data
-    uint64 FinishTimeUs = 109; // time in finished state // ComputeCpuTimeUs + PendingInputTimeUs + PendingOutputTimeUs + FinishTimeUs == 100% (or == const in aggregated graphs for several stages/tasks)
+    reserved 107;
+    reserved 108;
+    reserved 109;
 }
 
 message TDqComputeActorStats {

ydb/library/yql/dq/actors/task_runner/task_runner_actor_local.cpp

@@ -83,7 +83,6 @@ public:
 
 private:
     void OnStatisticsRequest(TEvStatistics::TPtr& ev) {
-        TaskRunner->UpdateStats();
 
         THashMap<ui32, const IDqAsyncOutputBuffer*> sinks;
         for (const auto sinkId : ev->Get()->SinkIds) {
@@ -220,7 +219,6 @@ private:
         {
             auto st = MakeHolder<TEvStatistics>(std::move(ev->Get()->SinkIds), std::move(ev->Get()->InputTransformIds));
 
-            TaskRunner->UpdateStats();
             THashMap<ui32, const IDqAsyncOutputBuffer*> sinks;
             for (const auto sinkId : st->SinkIds) {
                 sinks[sinkId] = TaskRunner->GetSink(sinkId).Get();