Fix empty upsert with sink (#17482)

Author: nikvas0

ydb/core/kqp/runtime/kqp_write_actor.cpp

@@ -1747,7 +1747,6 @@ class TKqpBufferWriteActor :public TActorBootstrapped<TKqpBufferWriteActor>, pub
 
     void Handle(TEvBufferWrite::TPtr& ev) {
         Counters->ForwardActorWritesLatencyHistogram->Collect((TInstant::Now() - ev->Get()->SendTime).MicroSeconds());
-
         TWriteToken token;
         if (!ev->Get()->Token) {
             AFL_ENSURE(ev->Get()->Settings);
@@ -2967,6 +2966,12 @@ class TKqpForwardWriteActor : public TActorBootstrapped<TKqpForwardWriteActor>,
 
     void Handle(TEvBufferWriteResult::TPtr& result) {
         CA_LOG_D("TKqpForwardWriteActor recieve EvBufferWriteResult from " << BufferActorId);
+
+        WriteToken = result->Get()->Token;
+        OnFlushed();
+    }
+
+    void OnFlushed() {
         InFlight = false;
 
         EgressStats.Bytes += DataSize;
@@ -2975,8 +2980,6 @@ class TKqpForwardWriteActor : public TActorBootstrapped<TKqpForwardWriteActor>,
         EgressStats.Resume();
 
         Counters->ForwardActorWritesSizeHistogram->Collect(DataSize);
-
-        WriteToken = result->Get()->Token;
         DataSize = 0;
 
         if (Closed) {
@@ -3032,6 +3035,12 @@ class TKqpForwardWriteActor : public TActorBootstrapped<TKqpForwardWriteActor>,
 
         ev->SendTime = TInstant::Now();
 
+        if (ev->Data->IsEmpty() && ev->Close && WriteToken.IsEmpty()) {
+            // Nothing was written
+            OnFlushed();
+            return;
+        }
+
         CA_LOG_D("Send data=" << DataSize << ", closed=" << Closed << ", bufferActorId=" << BufferActorId);
         AFL_ENSURE(Send(BufferActorId, ev.release()));
     }

ydb/core/kqp/ut/query/kqp_query_ut.cpp

@@ -2580,6 +2580,60 @@ Y_UNIT_TEST_SUITE(KqpQuery) {
         }
     }
 
+    Y_UNIT_TEST_TWIN(UpdateThenDelete, UseSink) {
+        NKikimrConfig::TAppConfig appConfig;
+        appConfig.MutableTableServiceConfig()->SetEnableOltpSink(UseSink);
+        auto settings = TKikimrSettings()
+            .SetAppConfig(appConfig)
+            .SetWithSampleTables(true);
+
+        TKikimrRunner kikimr(settings);
+        auto client = kikimr.GetTableClient();
+
+        {
+            const TString query = R"(
+                DECLARE $data AS List<Struct<
+                    Key: String,
+                    Value: String
+                >>;
+
+                UPSERT INTO KeyValue2 SELECT * FROM AS_TABLE($data);
+
+                DELETE FROM KeyValue2 ON SELECT * FROM KeyValue2 AS a LEFT ONLY JOIN AS_TABLE($data) AS b USING (Key);
+            )";
+
+            TTypeBuilder builder;
+            builder
+                .BeginStruct()
+                    .AddMember("Key", TTypeBuilder().Primitive(NYdb::EPrimitiveType::String).Build())
+                    .AddMember("Value", TTypeBuilder().Primitive(NYdb::EPrimitiveType::String).Build())
+                .EndStruct();
+
+            auto params = client.GetParamsBuilder()
+                .AddParam("$data")
+                    .EmptyList(builder.Build())
+                    .Build()
+                .Build();
+
+            auto session = client.CreateSession().GetValueSync().GetSession();
+            auto result = session.ExecuteDataQuery(query, NYdb::NTable::TTxControl::BeginTx().CommitTx(), std::move(params)).ExtractValueSync();
+            UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+
+        }
+        {
+            const TString query = R"(
+                SELECT * FROM KeyValue2;
+            )";
+
+            auto session = client.CreateSession().GetValueSync().GetSession();
+            auto result = session.ExecuteDataQuery(query, NYdb::NTable::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
+            UNIT_ASSERT_C(result.GetStatus() == NYdb::EStatus::SUCCESS, result.GetIssues().ToString());
+
+            Cerr << FormatResultSetYson(result.GetResultSet(0)) << Endl;
+            UNIT_ASSERT_VALUES_EQUAL(0, result.GetResultSet(0).RowsCount());
+        }
+    }
+
 }
 
 } // namespace NKqp