ut for deleting absent rows (#13172)

Author: zverevgeny

.github/config/muted_ya.txt

@@ -25,6 +25,7 @@ ydb/core/keyvalue/ut_trace TKeyValueTracingTest.WriteSmall
 ydb/core/kqp/ut/cost KqpCost.OlapWriteRow
 ydb/core/kqp/ut/data_integrity KqpDataIntegrityTrails.Select
 ydb/core/kqp/ut/data_integrity KqpDataIntegrityTrails.UpsertEvWrite
+ydb/core/kqp/ut/olap KqpOlap.DeleteAbsent+Reboot
 ydb/core/kqp/ut/olap KqpDecimalColumnShard.TestAggregation
 ydb/core/kqp/ut/olap KqpDecimalColumnShard.TestFilterCompare
 ydb/core/kqp/ut/olap KqpOlap.ManyColumnShardsWithRestarts

ydb/core/kqp/ut/olap/kqp_olap_ut.cpp

@@ -2991,6 +2991,36 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
                 .GetValueSync();
         UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SUCCESS);
     }
+
+    Y_UNIT_TEST_TWIN(DeleteAbsent, Reboot) {
+        //This test tries to DELETE from a table with WHERE condition that matches no rows
+        //It corresponds to a SCAN, then NO write then COMMIT
+        auto csController = NYDBTest::TControllers::RegisterCSControllerGuard<NYDBTest::NColumnShard::TController>();
+
+        NKikimrConfig::TAppConfig appConfig;
+        auto settings = TKikimrSettings().SetAppConfig(appConfig).SetWithSampleTables(false);
+        TTestHelper testHelper(settings);
+
+        TVector<TTestHelper::TColumnSchema> schema = {
+            TTestHelper::TColumnSchema().SetName("id").SetType(NScheme::NTypeIds::Int64).SetNullable(false),
+            TTestHelper::TColumnSchema().SetName("value").SetType(NScheme::NTypeIds::Int32).SetNullable(true),
+        };
+        TTestHelper::TColumnTable testTable;
+        testTable.SetName("/Root/ttt").SetPrimaryKey({ "id" }).SetSharding({ "id" }).SetSchema(schema);
+        testHelper.CreateTable(testTable);
+
+        if (Reboot) {
+            csController->SetRestartOnLocalTxCommitted("TProposeWriteTransaction");
+        }
+        auto client = testHelper.GetKikimr().GetQueryClient();
+        const auto resultDelete =
+            client
+                .ExecuteQuery(
+                    "DELETE from `/Root/ttt` WHERE value % 2 == 1;",
+                    NYdb::NQuery::TTxControl::BeginTx().CommitTx())
+                .GetValueSync();
+        UNIT_ASSERT_C(resultDelete.IsSuccess(), resultDelete.GetIssues().ToString());
+    }
 }
 
 }

ydb/core/tx/columnshard/columnshard__write.cpp

@@ -376,19 +376,19 @@ class TCommitOperation {
     ui64 ArbiterColumnShard = 0;
 };
 
-class TProposeWriteTransaction: public NTabletFlatExecutor::TTransactionBase<TColumnShard> {
+class TProposeWriteTransaction: public TExtendedTransactionBase {
 private:
-    using TBase = NTabletFlatExecutor::TTransactionBase<TColumnShard>;
+    using TBase = TExtendedTransactionBase;
 
 public:
     TProposeWriteTransaction(TColumnShard* self, TCommitOperation::TPtr op, const TActorId source, const ui64 cookie)
-        : TBase(self)
+        : TBase(self, "TProposeWriteTransaction")
         , WriteCommit(op)
         , Source(source)
         , Cookie(cookie) {
     }
 
-    virtual bool Execute(TTransactionContext& txc, const TActorContext&) override {
+    virtual bool DoExecute(TTransactionContext& txc, const TActorContext&) override {
         NKikimrTxColumnShard::TCommitWriteTxBody proto;
         NKikimrTxColumnShard::ETransactionKind kind;
         if (WriteCommit->NeedSyncLocks()) {
@@ -407,7 +407,7 @@ class TProposeWriteTransaction: public NTabletFlatExecutor::TTransactionBase<TCo
         return true;
     }
 
-    virtual void Complete(const TActorContext& ctx) override {
+    virtual void DoComplete(const TActorContext& ctx) override {
         Self->GetProgressTxController().FinishProposeOnComplete(WriteCommit->GetTxId(), ctx);
     }
     TTxType GetTxType() const override {

ydb/core/tx/columnshard/hooks/abstract/abstract.h

@@ -322,6 +322,13 @@ class ICSController {
     virtual void OnCleanupActors(const ui64 tabletId) {
         Y_UNUSED(tabletId);
     }
+
+    virtual void OnAfterLocalTxCommitted(const NActors::TActorContext& ctx, const NColumnShard::TColumnShard& shard, const TString& txInfo) {
+        Y_UNUSED(ctx);
+        Y_UNUSED(shard);
+        Y_UNUSED(txInfo);
+    }
+
 };
 
 class TControllers {

ydb/core/tx/columnshard/hooks/testing/controller.cpp

@@ -145,4 +145,10 @@ ::NKikimr::NColumnShard::TBlobPutResult::TPtr TController::OverrideBlobPutResult
     return result;
 }
 
+void TController::OnAfterLocalTxCommitted(const NActors::TActorContext& ctx, const ::NKikimr::NColumnShard::TColumnShard& shard, const TString& txInfo) {
+    if (RestartOnLocalDbTxCommitted == txInfo) {
+        ctx.Send(shard.SelfId(), new TEvents::TEvPoisonPill{});
+    }
+}
+
 }   // namespace NKikimr::NYDBTest::NColumnShard

ydb/core/tx/columnshard/hooks/testing/controller.h

@@ -132,6 +132,8 @@ class TController: public TReadOnlyController {
     void CheckInvariants(const ::NKikimr::NColumnShard::TColumnShard& shard, TCheckContext& context) const;
 
     THashSet<TString> SharingIds;
+
+    std::optional<TString> RestartOnLocalDbTxCommitted;
 protected:
     virtual const NOlap::NSplitter::TSplitSettings& DoGetBlobSplitSettings(const NOlap::NSplitter::TSplitSettings& defaultValue) const override {
         if (OverrideBlobSplitSettings) {
@@ -281,6 +283,13 @@ class TController: public TReadOnlyController {
         TGuard<TMutex> g(ActiveTabletsMutex);
         return ActiveTablets.contains(tabletId);
     }
+
+    void SetRestartOnLocalTxCommitted(std::optional<TString> txInfo) {
+        RestartOnLocalDbTxCommitted = std::move(txInfo);
+    }
+
+    virtual void OnAfterLocalTxCommitted(const NActors::TActorContext& ctx, const ::NKikimr::NColumnShard::TColumnShard& shard, const TString& txInfo) override;
+
 };
 
 }

ydb/core/tx/columnshard/tablet/ext_tx_base.cpp

@@ -11,6 +11,7 @@ bool TExtendedTransactionBase::Execute(NTabletFlatExecutor::TTransactionContext&
 void TExtendedTransactionBase::Complete(const NActors::TActorContext& ctx) {
     NActors::TLogContextGuard logGuard = NActors::TLogContextBuilder::Build()("tablet_id", Self->TabletID())("local_tx_no", TabletTxNo)("method", "complete")("tx_info", TxInfo);
     DoComplete(ctx);
+    NYDBTest::TControllers::GetColumnShardController()->OnAfterLocalTxCommitted(ctx, *Self, TxInfo);
 }
 
 TExtendedTransactionBase::TExtendedTransactionBase(TColumnShard* self, const TString& txInfo)