Disabling the tx cache in tests for forced serialization/deserialization of txs

Author: azevaykin

ydb/core/tx/datashard/datashard_pipeline.cpp

@@ -59,7 +59,7 @@ bool TPipeline::Load(NIceDb::TNiceDb& db) {
         return false;
     }
 
-    Config.Validate();
+    Config.SetDefaultLimits();
     UtmostCompleteTx = LastCompleteTx;
 
     return true;
@@ -1335,14 +1335,17 @@ ui64 TPipeline::GetInactiveTxSize() const {
 
 bool TPipeline::SaveForPropose(TValidatedTx::TPtr tx) {
     Y_ABORT_UNLESS(tx && tx->GetTxId());
-    if (DataTxCache.size() <= Config.LimitDataTxCache) {
-        ui64 quota = tx->GetMemoryConsumption();
-        if (Self->TryCaptureTxCache(quota)) {
-            tx->SetTxCacheUsage(quota);
-            DataTxCache[tx->GetTxId()] = tx;
-            return true;
-        }
+
+    if (DataTxCache.size() >= Config.LimitDataTxCache)
+        return false;
+
+    ui64 quota = tx->GetMemoryConsumption();
+    if (Self->TryCaptureTxCache(quota)) {
+        tx->SetTxCacheUsage(quota);
+        DataTxCache[tx->GetTxId()] = tx;
+        return true;
     }
+
     return false;
 }
 

ydb/core/tx/datashard/datashard_pipeline.h

@@ -55,7 +55,7 @@ class TPipeline : TNonCopyable {
         bool DirtyImmediate() const { return Flags & EFlagsDirtyImmediate; }
         bool SoftUpdates() const { return Flags & (EFlagsForceOnlineRW|EFlagsDirtyOnline|EFlagsDirtyImmediate); }
 
-        void Validate() {
+        void SetDefaultLimits() {
             if (!LimitActiveTx)
                 LimitActiveTx = DefaultLimitActiveTx() ;
             if (!LimitDataTxCache)
@@ -78,14 +78,17 @@ class TPipeline : TNonCopyable {
             } else {
                 Flags &= ~(EFlagsForceOnlineRW | EFlagsDirtyOnline | EFlagsDirtyImmediate);
             }
-            if (cfg.GetNumActiveTx()) {
+            if (cfg.HasNumActiveTx()) {
                 LimitActiveTx = cfg.GetNumActiveTx();
+            } else {
+                LimitActiveTx = DefaultLimitActiveTx();
             }
-            if (cfg.GetDataTxCacheSize()) {
+
+            if (cfg.HasDataTxCacheSize()) {
                 LimitDataTxCache = cfg.GetDataTxCacheSize();
+            } else {
+                LimitDataTxCache = DefaultLimitDataTxCache();
             }
-
-            Validate();
         }
     };
 

ydb/core/tx/datashard/datashard_ut_write.cpp

@@ -196,8 +196,7 @@ Y_UNIT_TEST_SUITE(DataShardWrite) {
 
         Cout << "========= Wait for completed transaction =========\n";
         {
-            auto ev = runtime.GrabEdgeEventRethrow<NEvents::TDataEvents::TEvWriteResult>(sender);
-            auto writeResult = ev->Get()->Record;
+            auto writeResult = WaitForWriteCompleted(runtime, sender);
 
             UNIT_ASSERT_VALUES_EQUAL_C(writeResult.GetStatus(), NKikimrDataEvents::TEvWriteResult::STATUS_COMPLETED, "Status: " << writeResult.GetStatus() << " Issues: " << writeResult.GetIssues());
             UNIT_ASSERT_VALUES_EQUAL(writeResult.GetOrigin(), shard);
@@ -217,6 +216,43 @@ Y_UNIT_TEST_SUITE(DataShardWrite) {
             UNIT_ASSERT_VALUES_EQUAL(tableState, expectedTableState);
         }
 
+    }
+    
+    Y_UNIT_TEST(WritePreparedNoTxCache) {
+        auto [runtime, server, sender] = TestCreateServer();
+
+        TShardedTableOptions opts;
+        opts.DataTxCacheSize(0);  //disabling the tx cache for forced serialization/deserialization of txs
+        const auto [shards, tableId] = CreateShardedTable(server, sender, "/Root", "table-1", opts);
+        const ui64 shard = shards[0];
+        const ui32 rowCount = 3;
+
+        ui64 txId = 100;
+        ui64 minStep, maxStep;
+
+        Cout << "========= Send prepare =========\n";
+        {
+            const auto writeResult = Write(runtime, sender, shard, tableId, opts.Columns_, rowCount, txId, NKikimrDataEvents::TEvWrite::MODE_PREPARE);
+            minStep = writeResult.GetMinStep();
+            maxStep = writeResult.GetMaxStep();
+        }
+
+        Cout << "========= Send propose to coordinator =========\n";
+        {
+            SendProposeToCoordinator(server, shards, minStep, maxStep, txId);
+        }
+
+        Cout << "========= Wait for completed transaction =========\n";
+        {
+            WaitForWriteCompleted(runtime, sender);
+        }
+
+        Cout << "========= Read table =========\n";
+        {
+            auto tableState = TReadTableState(server, MakeReadTableSettings("/Root/table-1")).All();
+            UNIT_ASSERT_VALUES_EQUAL(tableState, expectedTableState);
+        }
+
     } // Y_UNIT_TEST
 
 } // Y_UNIT_TEST_SUITE

ydb/core/tx/datashard/ut_common/datashard_ut_common.cpp

@@ -1184,6 +1184,9 @@ std::tuple<TVector<ui64>, TTableId> CreateShardedTable(
     if (!opts.EnableOutOfOrder_)
         desc->MutablePartitionConfig()->MutablePipelineConfig()->SetEnableOutOfOrder(false);
 
+    if (opts.DataTxCacheSize_)
+        desc->MutablePartitionConfig()->MutablePipelineConfig()->SetDataTxCacheSize(*opts.DataTxCacheSize_);
+
     if (opts.Policy_) {
         opts.Policy_->Serialize(*desc->MutablePartitionConfig()->MutableCompactionPolicy());
     }
@@ -1986,6 +1989,14 @@ TTestActorRuntimeBase::TEventObserverHolderPair ReplaceEvProposeTransactionWithE
     return {std::move(requestObserver), std::move(responseObserver)};
 }
 
+NKikimrDataEvents::TEvWriteResult WaitForWriteCompleted(TTestActorRuntime& runtime, TActorId sender)
+{
+    auto ev = runtime.GrabEdgeEventRethrow<NEvents::TDataEvents::TEvWriteResult>(sender);
+    auto resultRecord = ev->Get()->Record;
+    UNIT_ASSERT_C(resultRecord.GetStatus() == NKikimrDataEvents::TEvWriteResult::STATUS_COMPLETED, "Status: " << resultRecord.GetStatus() << " Issues: " << resultRecord.GetIssues());
+    return resultRecord;
+}
+
 void UploadRows(TTestActorRuntime& runtime, const TString& tablePath, const TVector<std::pair<TString, Ydb::Type_PrimitiveTypeId>>& types, const TVector<TCell>& keys, const TVector<TCell>& values)
 {
     auto txTypes = std::make_shared<NTxProxy::TUploadTypes>();

ydb/core/tx/datashard/ut_common/datashard_ut_common.h

@@ -475,6 +475,7 @@ struct TShardedTableOptions {
     TABLE_OPTION(bool, FollowerPromotion, false);
     TABLE_OPTION(bool, ExternalStorage, false);
     TABLE_OPTION(std::optional<ui64>, ExecutorCacheSize, std::nullopt);
+    TABLE_OPTION(std::optional<ui32>, DataTxCacheSize, std::nullopt);
     TABLE_OPTION(bool, Replicated, false);
     TABLE_OPTION(std::optional<EReplicationConsistency>, ReplicationConsistency, std::nullopt);
     TABLE_OPTION(TAttributes, Attributes, {});
@@ -710,6 +711,7 @@ void ExecSQL(Tests::TServer::TPtr server,
 
 NKikimrDataEvents::TEvWriteResult Write(TTestActorRuntime& runtime, TActorId sender, ui64 shardId, std::unique_ptr<NEvents::TDataEvents::TEvWrite>&& request, NKikimrDataEvents::TEvWriteResult::EStatus expectedStatus = NKikimrDataEvents::TEvWriteResult::STATUS_UNSPECIFIED, NWilson::TTraceId traceId = {});
 NKikimrDataEvents::TEvWriteResult Write(TTestActorRuntime& runtime, TActorId sender, ui64 shardId, const TTableId& tableId, const TVector<TShardedTableOptions::TColumn>& columns, ui32 rowCount, ui64 txId, NKikimrDataEvents::TEvWrite::ETxMode txMode, NKikimrDataEvents::TEvWriteResult::EStatus expectedStatus = NKikimrDataEvents::TEvWriteResult::STATUS_UNSPECIFIED, NWilson::TTraceId traceId = {});
+NKikimrDataEvents::TEvWriteResult WaitForWriteCompleted(TTestActorRuntime& runtime, TActorId sender); 
 
 struct TEvWriteRow {
     TEvWriteRow(const TTableId& tableId, std::initializer_list<ui32> init)