Merge 79a09f7 into cd0aab0

Author: GrigoriyPA

ydb/core/kqp/executer_actor/kqp_data_executer.cpp

@@ -2834,7 +2834,9 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
             << ", execution id " << context->CurrentExecutionId
             << ", checkpoint id " << checkpointId << ", generation " << Generation
             << ", state load mode " << FederatedQuery::StateLoadMode_Name(stateLoadMode)
-            << ", streaming disposition " << streamingDisposition.ShortDebugString());
+            << ", streaming disposition " << streamingDisposition.ShortDebugString()
+            << ", has QueryPhysicalGraph " << (Request.QueryPhysicalGraph != nullptr)
+            << ", enabled watermarks " << Request.QueryPhysicalGraph->GetPreparedQuery().GetPhysicalQuery().GetEnableWatermarks());
     }
 
 private:

ydb/core/kqp/executer_actor/kqp_planner.cpp

@@ -639,6 +639,7 @@ void TKqpPlanner::PrepareCheckpoints() {
     }
 
     const auto enableCheckpoints = static_cast<bool>(CheckpointCoordinatorId);
+    LOG_D("BuildCheckpointingAndWatermarksMode, enableCheckpoints: " << enableCheckpoints << ", EnableWatermarks: " << EnableWatermarks);
     TasksGraph.BuildCheckpointingAndWatermarksMode(enableCheckpoints, EnableWatermarks);
 
     if (!enableCheckpoints) {

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

@@ -1555,9 +1555,11 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
         auto sink = outputInfo.Buffer;
 
         NKikimr::NMiniKQL::TUnboxedValueBatch dataBatch(sink->GetOutputType());
+        NDqProto::TWatermark watermark;
         NDqProto::TCheckpoint checkpoint;
 
         const ui64 dataSize = !outputInfo.Finished ? sink->Pop(dataBatch, bytes) : 0;
+        std::ignore = sink->Pop(watermark);
         const bool hasCheckpoint = sink->Pop(checkpoint);
         if (!dataSize && !hasCheckpoint) {
             if (!sink->IsFinished()) {

ydb/library/yql/dq/runtime/dq_input_producer.cpp

@@ -76,7 +76,7 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
 
         if (Batch.empty()) {
             // pass watermark and wait for drain only if watermarks enabled
-            if (WatermarksEnabled() && TrySendWatermark()) {
+            if (TrySendWatermark()) {
                 return NUdf::EFetchStatus::Yield;
             }
 
@@ -94,7 +94,8 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
             }
 
             // pass watermark and wait for drain only if watermarks enabled and batch is still empty
-            if (Batch.empty() && WatermarksEnabled() && TrySendWatermark()) {
+            if (Batch.empty()) {
+                std::ignore = TrySendWatermark();
                 return NUdf::EFetchStatus::Yield;
             }
         }
@@ -122,7 +123,7 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
 
         if (Batch.empty()) {
             // pass watermark and wait for drain only if watermarks enabled
-            if (WatermarksEnabled() && TrySendWatermark()) {
+            if (TrySendWatermark()) {
                 return NUdf::EFetchStatus::Yield;
             }
 
@@ -140,7 +141,8 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
             }
 
             // pass watermark and wait for drain only if watermarks enabled and batch is still empty
-            if (Batch.empty() && WatermarksEnabled() && TrySendWatermark()) {
+            if (Batch.empty()) {
+                std::ignore = TrySendWatermark();
                 return NUdf::EFetchStatus::Yield;
             }
         }
@@ -200,8 +202,10 @@ class TDqInputUnionStreamValue : public TComputationValue<TDqInputUnionStreamVal
     }
 
     [[nodiscard]] bool TrySendWatermark() {
-        Y_DEBUG_ABORT_UNLESS(WatermarksEnabled());
-        if (!Watermark || !NotifyWatermarkTracker(InputKey, *Watermark)) {
+        if (!WatermarksEnabled()) {
+            return false;
+        }
+        if (!Watermark || !NotifyWatermarkTracker(InputKey, *Watermark) || !WatermarksTracker->HasPendingWatermark()) {
             return false;
         }
         Y_DEBUG_ABORT_UNLESS(WatermarksTracker->HasPendingWatermark());

yql/essentials/minikql/comp_nodes/mkql_multihopping.cpp

@@ -408,7 +408,7 @@ class TMultiHoppingCoreWrapper: public TStatefulSourceComputationNode<TMultiHopp
             auto curHopIndex = keyState.HopIndex;
             auto curHopIndexModBuckets = curHopIndex % bucketsForKey.size();
 
-            if (curHopIndex > closeBeforeIndex) {
+            if (curHopIndex >= closeBeforeIndex) {
                 return keyState.NextHopIndex <= keyState.HopIndex && keyState.FutureEvents.empty();
             }