Accurate TR timings (#14487)

Hor911 · blinkov · commit e8267c38f714 · 2025-03-21T09:51:52.000+07:00
diff --git a/ydb/library/yql/dq/actors/compute/dq_compute_actor_stats.cpp b/ydb/library/yql/dq/actors/compute/dq_compute_actor_stats.cpp
@@ -284,8 +284,9 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TTaskRunnerStatsBase&
     protoTask->SetResultRows(resultStats.Rows);
     protoTask->SetResultBytes(resultStats.Bytes);
 
-    protoTask->SetFinishTimeMs(finishTime.MilliSeconds());
+    protoTask->SetCreateTimeMs(taskStats.CreateTs.MilliSeconds());
     protoTask->SetStartTimeMs(startTime.MilliSeconds());
+    protoTask->SetFinishTimeMs(finishTime.MilliSeconds());
 }
 
 } // namespace NDq
diff --git a/ydb/library/yql/dq/actors/protos/dq_stats.proto b/ydb/library/yql/dq/actors/protos/dq_stats.proto
@@ -247,11 +247,8 @@ message TDqTaskStats {
 
     google.protobuf.Any Extra = 200;
 
-    // BASIC: StartTs from TR since we don't have better times
-    // FULL+: min(FirstMessageMs) from all Inputs
+    uint64 CreateTimeMs = 166;
     uint64 StartTimeMs = 158;
-
-    // always FinishTs from TR
     uint64 FinishTimeMs = 5;
 
     reserved 4;   // was FirstRowTimeMs
diff --git a/ydb/library/yql/dq/runtime/dq_input_producer.cpp b/ydb/library/yql/dq/runtime/dq_input_producer.cpp
@@ -24,12 +24,15 @@ template<bool IsWide>
 class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamValue<IsWide>> {
     using TBase = TComputationValue<TDqInputUnionStreamValue<IsWide>>;
 public:
-    TDqInputUnionStreamValue(TMemoryUsageInfo* memInfo, const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs, TDqMeteringStats::TInputStatsMeter stats)
+    TDqInputUnionStreamValue(TMemoryUsageInfo* memInfo, const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
+        TDqMeteringStats::TInputStatsMeter stats, TInstant& startTs, bool& inputConsumed)
         : TBase(memInfo)
         , Inputs(std::move(inputs))
         , Alive(Inputs.size())
         , Batch(type)
         , Stats(stats)
+        , StartTs(startTs)
+        , InputConsumed(inputConsumed)
     {}
 
 private:
@@ -41,6 +44,10 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
                 case NUdf::EFetchStatus::Ok:
                     break;
                 case NUdf::EFetchStatus::Finish:
+                    if (Y_UNLIKELY(!StartTs)) {
+                        StartTs = Now();
+                    }
+                    [[fallthrough]];
                 case NUdf::EFetchStatus::Yield:
                     return status;
             }
@@ -52,6 +59,10 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
         if (Stats) {
             Stats.Add(result);
         }
+        if (Y_UNLIKELY(!StartTs)) {
+            StartTs = Now();
+        }
+        InputConsumed = true;
         return NUdf::EFetchStatus::Ok;
     }
 
@@ -63,6 +74,10 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
                 case NUdf::EFetchStatus::Ok:
                     break;
                 case NUdf::EFetchStatus::Finish:
+                    if (Y_UNLIKELY(!StartTs)) {
+                        StartTs = Now();
+                    }
+                    [[fallthrough]];
                 case NUdf::EFetchStatus::Yield:
                     return status;
             }
@@ -76,6 +91,10 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
         if (Stats) {
             Stats.Add(result, width);
         }
+        if (Y_UNLIKELY(!StartTs)) {
+            StartTs = Now();
+        }
+        InputConsumed = true;
         return NUdf::EFetchStatus::Ok;
     }
 
@@ -108,19 +127,23 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
     size_t Index = 0;
     TUnboxedValueBatch Batch;
     TDqMeteringStats::TInputStatsMeter Stats;
+    TInstant& StartTs;
+    bool& InputConsumed;
 };
 
 template<bool IsWide>
 class TDqInputMergeStreamValue : public TComputationValue<TDqInputMergeStreamValue<IsWide>> {
     using TBase = TComputationValue<TDqInputMergeStreamValue<IsWide>>;
 public:
     TDqInputMergeStreamValue(TMemoryUsageInfo* memInfo, const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
-        TVector<TSortColumnInfo>&& sortCols, TDqMeteringStats::TInputStatsMeter stats)
+        TVector<TSortColumnInfo>&& sortCols, TDqMeteringStats::TInputStatsMeter stats, TInstant& startTs, bool& inputConsumed)
         : TBase(memInfo)
         , Inputs(std::move(inputs))
         , Width(type->IsMulti() ? static_cast<const NMiniKQL::TMultiType*>(type)->GetElementsCount() : TMaybe<ui32>())
         , SortCols(std::move(sortCols))
         , Stats(stats)
+        , StartTs(startTs)
+        , InputConsumed(inputConsumed)
     {
         YQL_ENSURE(!IsWide ^ Width.Defined());
         CurrentBuffers.reserve(Inputs.size());
@@ -196,6 +219,10 @@ class TDqInputMergeStreamValue : public TComputationValue<TDqInputMergeStreamVal
             case NUdf::EFetchStatus::Ok:
                 break;
             case NUdf::EFetchStatus::Finish:
+                if (Y_UNLIKELY(!StartTs)) {
+                    StartTs = Now();
+                }
+                [[fallthrough]];
             case NUdf::EFetchStatus::Yield:
                 return status;
         }
@@ -204,6 +231,10 @@ class TDqInputMergeStreamValue : public TComputationValue<TDqInputMergeStreamVal
         if (Stats) {
             Stats.Add(result);
         }
+        if (Y_UNLIKELY(!StartTs)) {
+            StartTs = Now();
+        }
+        InputConsumed = true;
         return NUdf::EFetchStatus::Ok;
     }
 
@@ -223,6 +254,7 @@ class TDqInputMergeStreamValue : public TComputationValue<TDqInputMergeStreamVal
         if (Stats) {
             Stats.Add(result, width);
         }
+        InputConsumed = true;
         return NUdf::EFetchStatus::Ok;
     }
 
@@ -309,6 +341,8 @@ class TDqInputMergeStreamValue : public TComputationValue<TDqInputMergeStreamVal
     ui32 InitializationIndex = 0;
     TMap<ui32, EDataSlot> SortColTypes;
     TDqMeteringStats::TInputStatsMeter Stats;
+    TInstant& StartTs;
+    bool& InputConsumed;
 };
 
 TVector<NKikimr::NMiniKQL::TType*> ExtractBlockItemTypes(const NKikimr::NMiniKQL::TType* type) {
@@ -380,7 +414,7 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
 public:
     TDqInputMergeBlockStreamValue(TMemoryUsageInfo* memInfo, const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
         TVector<TSortColumnInfo>&& sortCols, const NKikimr::NMiniKQL::THolderFactory& factory, TDqMeteringStats::TInputStatsMeter stats,
-        NUdf::IPgBuilder* pgBuilder)
+        TInstant& startTs, bool& inputConsumed, NUdf::IPgBuilder* pgBuilder)
         : TBase(memInfo)
         , SortCols_(std::move(sortCols))
         , ItemTypes_(ExtractBlockItemTypes(type))
@@ -389,6 +423,8 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
         , Builders_(MakeBuilders(MaxOutputBlockLen_, ItemTypes_, pgBuilder))
         , Factory_(factory)
         , Stats_(stats)
+        , StartTs(startTs)
+        , InputConsumed(inputConsumed)
     {
         YQL_ENSURE(MaxOutputBlockLen_ > 0);
         InputData_.reserve(inputs.size());
@@ -552,6 +588,9 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
     NUdf::EFetchStatus WideFetch(NKikimr::NUdf::TUnboxedValue* result, ui32 width) final {
         YQL_ENSURE(width == ItemTypes_.size() + 1);
         if (IsFinished_) {
+            if (Y_UNLIKELY(!StartTs)) {
+                StartTs = Now();
+            }
             return NUdf::EFetchStatus::Finish;
         }
 
@@ -576,6 +615,10 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
         if (Stats_) {
             Stats_.Add(result, width);
         }
+        if (Y_UNLIKELY(!StartTs)) {
+            StartTs = Now();
+        }
+        InputConsumed = true;
         return NUdf::EFetchStatus::Ok;
     }
 
@@ -645,7 +688,7 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
                 if (status == NUdf::EFetchStatus::Ok) {
                     std::make_heap(InputRows_.begin(), InputRows_.end());
                 }
-            }            
+            }
         }
 
         if (!OutputBlockLen_) {
@@ -685,9 +728,11 @@ class TDqInputMergeBlockStreamValue : public TComputationValue<TDqInputMergeBloc
 
     const NKikimr::NMiniKQL::THolderFactory& Factory_;
     TDqMeteringStats::TInputStatsMeter Stats_;
-    
+
     std::unique_ptr<TArgsDechunker> Output_;
     bool IsFinished_ = false;
+    TInstant& StartTs;
+    bool& InputConsumed;
 };
 
 void ValidateInputTypes(const NKikimr::NMiniKQL::TType* type, const TVector<IDqInput::TPtr>& inputs) {
@@ -723,7 +768,7 @@ void TDqMeteringStats::TInputStatsMeter::Add(const NKikimr::NUdf::TUnboxedValue*
         } else {
             Stats->RowsConsumed += 1;
         }
-        
+
         NYql::NDq::TDqDataSerializer::TEstimateSizeSettings settings;
         settings.DiscardUnsupportedTypes = true;
         settings.WithHeaders = false;
@@ -740,18 +785,18 @@ void TDqMeteringStats::TInputStatsMeter::Add(const NKikimr::NUdf::TUnboxedValue*
 }
 
 NUdf::TUnboxedValue CreateInputUnionValue(const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
-    const NMiniKQL::THolderFactory& factory, TDqMeteringStats::TInputStatsMeter stats)
+    const NMiniKQL::THolderFactory& factory, TDqMeteringStats::TInputStatsMeter stats, TInstant& startTs, bool& inputConsumed)
 {
     ValidateInputTypes(type, inputs);
     if (type->IsMulti()) {
-        return factory.Create<TDqInputUnionStreamValue<true>>(type, std::move(inputs), stats);
+        return factory.Create<TDqInputUnionStreamValue<true>>(type, std::move(inputs), stats, startTs, inputConsumed);
     }
-    return factory.Create<TDqInputUnionStreamValue<false>>(type, std::move(inputs), stats);
+    return factory.Create<TDqInputUnionStreamValue<false>>(type, std::move(inputs), stats, startTs, inputConsumed);
 }
 
 NKikimr::NUdf::TUnboxedValue CreateInputMergeValue(const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
     TVector<TSortColumnInfo>&& sortCols, const NKikimr::NMiniKQL::THolderFactory& factory, TDqMeteringStats::TInputStatsMeter stats,
-    NUdf::IPgBuilder* pgBuilder)
+    TInstant& startTs, bool& inputConsumed, NUdf::IPgBuilder* pgBuilder)
 {
     ValidateInputTypes(type, inputs);
     YQL_ENSURE(!inputs.empty());
@@ -760,13 +805,13 @@ NKikimr::NUdf::TUnboxedValue CreateInputMergeValue(const NKikimr::NMiniKQL::TTyp
             // we can ignore scalar columns, since all they have exactly the same value in all inputs
             EraseIf(sortCols, [](const auto& sortCol) { return *sortCol.IsScalar; });
             if (sortCols.empty()) {
-                return factory.Create<TDqInputUnionStreamValue<true>>(type, std::move(inputs), stats);
+                return factory.Create<TDqInputUnionStreamValue<true>>(type, std::move(inputs), stats, startTs, inputConsumed);
             }
-            return factory.Create<TDqInputMergeBlockStreamValue>(type, std::move(inputs), std::move(sortCols), factory, stats, pgBuilder);
+            return factory.Create<TDqInputMergeBlockStreamValue>(type, std::move(inputs), std::move(sortCols), factory, stats, startTs, inputConsumed, pgBuilder);
         }
-        return factory.Create<TDqInputMergeStreamValue<true>>(type, std::move(inputs), std::move(sortCols), stats);
+        return factory.Create<TDqInputMergeStreamValue<true>>(type, std::move(inputs), std::move(sortCols), stats, startTs, inputConsumed);
     }
-    return factory.Create<TDqInputMergeStreamValue<false>>(type, std::move(inputs), std::move(sortCols), stats);
+    return factory.Create<TDqInputMergeStreamValue<false>>(type, std::move(inputs), std::move(sortCols), stats, startTs, inputConsumed);
 }
 
 } // namespace NYql::NDq
diff --git a/ydb/library/yql/dq/runtime/dq_input_producer.h b/ydb/library/yql/dq/runtime/dq_input_producer.h
@@ -31,10 +31,10 @@ struct TDqMeteringStats {
 };
 
 NKikimr::NUdf::TUnboxedValue CreateInputUnionValue(const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
-    const NKikimr::NMiniKQL::THolderFactory& holderFactory, TDqMeteringStats::TInputStatsMeter = {});
+    const NKikimr::NMiniKQL::THolderFactory& holderFactory, TDqMeteringStats::TInputStatsMeter, TInstant& startTs, bool& inputConsumed);
 
 NKikimr::NUdf::TUnboxedValue CreateInputMergeValue(const NKikimr::NMiniKQL::TType* type, TVector<IDqInput::TPtr>&& inputs,
     TVector<TSortColumnInfo>&& sortCols, const NKikimr::NMiniKQL::THolderFactory& factory,
-    TDqMeteringStats::TInputStatsMeter = {}, NUdf::IPgBuilder* pgBuilder = nullptr);
+    TDqMeteringStats::TInputStatsMeter, TInstant& startTs, bool& inputConsumed, NUdf::IPgBuilder* pgBuilder = nullptr);
 
 } // namespace NYql::NDq
diff --git a/ydb/library/yql/dq/runtime/dq_tasks_runner.cpp b/ydb/library/yql/dq/runtime/dq_tasks_runner.cpp
@@ -143,21 +143,21 @@ void ValidateParamValue(std::string_view paramName, const TType* type, const NUd
 
 NUdf::TUnboxedValue DqBuildInputValue(const NDqProto::TTaskInput& inputDesc, const NKikimr::NMiniKQL::TType* type,
     TVector<IDqInput::TPtr>&& inputs, const THolderFactory& holderFactory, TDqMeteringStats::TInputStatsMeter stats,
-    NUdf::IPgBuilder* pgBuilder)
+    TInstant& startTs, bool& inputConsumed, NUdf::IPgBuilder* pgBuilder)
 {
     switch (inputDesc.GetTypeCase()) {
         case NYql::NDqProto::TTaskInput::kSource:
             Y_ABORT_UNLESS(inputs.size() == 1);
             [[fallthrough]];
         case NYql::NDqProto::TTaskInput::kUnionAll:
-            return CreateInputUnionValue(type, std::move(inputs), holderFactory, stats);
+            return CreateInputUnionValue(type, std::move(inputs), holderFactory, stats, startTs, inputConsumed);
         case NYql::NDqProto::TTaskInput::kMerge: {
             const auto& protoSortCols = inputDesc.GetMerge().GetSortColumns();
             TVector<TSortColumnInfo> sortColsInfo;
             GetSortColumnsInfo(type, protoSortCols, sortColsInfo);
             YQL_ENSURE(!sortColsInfo.empty());
 
-            return CreateInputMergeValue(type, std::move(inputs), std::move(sortColsInfo), holderFactory, stats, pgBuilder);
+            return CreateInputMergeValue(type, std::move(inputs), std::move(sortColsInfo), holderFactory, stats, startTs, inputConsumed, pgBuilder);
         }
         default:
             YQL_ENSURE(false, "Unknown input type: " << (ui32) inputDesc.GetTypeCase());
@@ -240,12 +240,10 @@ class TDqTaskRunner : public IDqTaskRunner {
         , LogFunc(logFunc)
         , AllocatedHolder(std::make_optional<TAllocatedHolder>())
     {
-        if (CollectBasic()) {
-            Stats = std::make_unique<TDqTaskRunnerStats>();
-            Stats->StartTs = TInstant::Now();
-            if (Y_UNLIKELY(CollectFull())) {
-                Stats->ComputeCpuTimeByRun = NMonitoring::ExponentialHistogram(6, 10, 10);
-            }
+        Stats = std::make_unique<TDqTaskRunnerStats>();
+        Stats->CreateTs = TInstant::Now();
+        if (Y_UNLIKELY(CollectFull())) {
+            Stats->ComputeCpuTimeByRun = NMonitoring::ExponentialHistogram(6, 10, 10);
         }
 
         if (Context.TypeEnv) {
@@ -531,7 +529,7 @@ class TDqTaskRunner : public IDqTaskRunner {
 
         for (ui32 i = 0; i < task.InputsSize(); ++i) {
             auto& inputDesc = task.GetInputs(i);
-            TDqMeteringStats::TInputStats* inputStats = nullptr; 
+            TDqMeteringStats::TInputStats* inputStats = nullptr;
             if (task.EnableMetering()) {
                 inputStats = &MeteringStats.AddInputs();
             }
@@ -592,16 +590,16 @@ class TDqTaskRunner : public IDqTaskRunner {
 
             auto entryNode = AllocatedHolder->ProgramParsed.CompGraph->GetEntryPoint(i, true);
             if (transform) {
-                transform->TransformInput = DqBuildInputValue(inputDesc, transform->TransformInputType, std::move(inputs), holderFactory, {}, PgBuilder_.get());
+                transform->TransformInput = DqBuildInputValue(inputDesc, transform->TransformInputType, std::move(inputs), holderFactory, {}, Stats->StartTs, InputConsumed, PgBuilder_.get());
                 inputs.clear();
                 inputs.emplace_back(transform->TransformOutput);
                 entryNode->SetValue(AllocatedHolder->ProgramParsed.CompGraph->GetContext(),
                     CreateInputUnionValue(transform->TransformOutput->GetInputType(), std::move(inputs), holderFactory,
-                        {inputStats, transform->TransformOutputType}));
+                        {inputStats, transform->TransformOutputType}, Stats->StartTs, InputConsumed));
             } else {
                 entryNode->SetValue(AllocatedHolder->ProgramParsed.CompGraph->GetContext(),
                     DqBuildInputValue(inputDesc, entry->InputItemTypes[i], std::move(inputs), holderFactory,
-                        {inputStats, entry->InputItemTypes[i]}, PgBuilder_.get()));
+                        {inputStats, entry->InputItemTypes[i]}, Stats->StartTs, InputConsumed, PgBuilder_.get()));
             }
         }
 
@@ -733,6 +731,7 @@ class TDqTaskRunner : public IDqTaskRunner {
             StopWaiting();
         }
 
+        InputConsumed = false;
         auto runStatus = FetchAndDispatch();
 
         if (Y_UNLIKELY(CollectFull())) {
@@ -766,7 +765,9 @@ class TDqTaskRunner : public IDqTaskRunner {
                     Stats->FinishTs = TInstant::Now();
                     break;
                 case ERunStatus::PendingInput:
-                    StartWaitingInput();
+                    if (!InputConsumed) {
+                        StartWaitingInput();
+                    }
                     break;
                 case ERunStatus::PendingOutput:
                     StartWaitingOutput();
@@ -971,6 +972,7 @@ class TDqTaskRunner : public IDqTaskRunner {
     TLogFunc LogFunc;
     std::unique_ptr<NUdf::ISecureParamsProvider> SecureParamsProvider;
     TDqTaskCountersProvider CountersProvider;
+    bool InputConsumed = false;
 
     struct TInputTransformInfo {
         NUdf::TUnboxedValue TransformInput;
@@ -1042,7 +1044,11 @@ class TDqTaskRunner : public IDqTaskRunner {
 
     void StopWaiting() {
         if (StartWaitInputTime) {
-            Stats->WaitInputTime += (TInstant::Now() - *StartWaitInputTime);
+            if (Y_LIKELY(Stats->StartTs)) {
+                Stats->WaitInputTime += (TInstant::Now() - *StartWaitInputTime);
+            } else {
+                Stats->WaitStartTime += (TInstant::Now() - *StartWaitInputTime);
+            }
             StartWaitInputTime.reset();
         }
         if (StartWaitOutputTime) {
diff --git a/ydb/library/yql/dq/runtime/dq_tasks_runner.h b/ydb/library/yql/dq/runtime/dq_tasks_runner.h

Original file line number	Diff line number	Diff line change
`@@ -284,8 +284,9 @@ void FillTaskRunnerStats(ui64 taskId, ui32 stageId, const TTaskRunnerStatsBase&`
`284`	`284`	`protoTask->SetResultRows(resultStats.Rows);`
`285`	`285`	`protoTask->SetResultBytes(resultStats.Bytes);`
`286`	`286`
`287`		`- protoTask->SetFinishTimeMs(finishTime.MilliSeconds());`
	`287`	`+ protoTask->SetCreateTimeMs(taskStats.CreateTs.MilliSeconds());`
`288`	`288`	`protoTask->SetStartTimeMs(startTime.MilliSeconds());`
	`289`	`+ protoTask->SetFinishTimeMs(finishTime.MilliSeconds());`
`289`	`290`	`}`
`290`	`291`
`291`	`292`	`} // namespace NDq`