Fixed Column Shard write stats (ydb-platform#24736)

Author: neyrox

.github/config/muted_ya.txt

@@ -46,7 +46,6 @@ ydb/core/kqp/ut/olap KqpOlapSysView.StatsSysViewTable
 ydb/core/kqp/ut/opt KqpKv.ReadRows_TimeoutCancelsReads
 ydb/core/kqp/ut/query KqpAnalyze.AnalyzeTable+ColumnStore
 ydb/core/kqp/ut/query KqpLimits.ComputeNodeMemoryLimit
-ydb/core/kqp/ut/query KqpStats.CreateTableAsStats+IsOlap
 ydb/core/kqp/ut/query KqpStats.DeferredEffects+UseSink
 ydb/core/kqp/ut/scan KqpFlowControl.FlowControl_Unlimited
 ydb/core/kqp/ut/scan KqpScan.StreamExecuteScanQueryClientTimeoutBruteForce

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

@@ -915,11 +915,21 @@ Y_UNIT_TEST_TWIN(CreateTableAsStats, IsOlap) {
         auto stats = NYdb::TProtoAccessor::GetProto(*result.GetStats());
         Cerr << stats.DebugString() << Endl;
         UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(0).updates().rows(), 2);
-        UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(0).updates().bytes(), 24);
-        UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(0).partitions_count(), 1);
         UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).reads().rows(), 2);
-        UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).reads().bytes(), 24);
-        UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).partitions_count(), 1);
+
+        if (IsOlap) {
+            // size of serialized may be a little different (because of arrow)
+            UNIT_ASSERT_GE(stats.query_phases(0).table_access(0).updates().bytes(), 400);
+            UNIT_ASSERT_LE(stats.query_phases(0).table_access(0).updates().bytes(), 500);
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).reads().bytes(), 40);
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(0).partitions_count(), 2);
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).partitions_count(), 0);
+        } else {
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(0).updates().bytes(), 24);
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).reads().bytes(), 24);
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(0).partitions_count(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(0).table_access(1).partitions_count(), 1);
+        }
     }
 
     {

ydb/core/tx/columnshard/blobs_action/common/statistics.cpp

@@ -3,14 +3,17 @@
 namespace NKikimr::NColumnShard {
 
 void AddTableAccessStatsToTxStats(NKikimrQueryStats::TTxStats& stats, ui64 pathId, ui64 rows, ui64 bytes,
-                                  NEvWrite::EModificationType modificationType) {
+                                  NEvWrite::EModificationType modificationType, ui64 shardId) {
     auto tableStats = stats.AddTableAccessStats();
     tableStats->MutableTableInfo()->SetPathId(pathId);
     auto row = modificationType == NEvWrite::EModificationType::Delete ? tableStats->MutableEraseRow()
                                                                        : tableStats->MutableUpdateRow();
     row->SetCount(rows);
     row->SetRows(rows);
     row->SetBytes(bytes);
+
+    auto shardStats = stats.AddPerShardStats();
+    shardStats->SetShardId(shardId);
 }
 
 } // namespace NKikimr::NColumnShard

ydb/core/tx/columnshard/blobs_action/common/statistics.h

@@ -7,6 +7,6 @@
 namespace NKikimr::NColumnShard {
 
 void AddTableAccessStatsToTxStats(NKikimrQueryStats::TTxStats& stats, ui64 pathId, ui64 rows, ui64 bytes,
-                                  NEvWrite::EModificationType modificationType);
+                                  NEvWrite::EModificationType modificationType, ui64 shardId);
 
 } // namespace NKikimr::NColumnShard

ydb/core/tx/columnshard/blobs_action/transaction/tx_blobs_written.cpp

@@ -72,6 +72,8 @@ bool TTxBlobsWritingFinished::DoExecute(TTransactionContext& txc, const TActorCo
         if (operation->GetBehaviour() == EOperationBehaviour::NoTxWrite) {
             LWPROBE(EvWriteResult, Self->TabletID(), writeMeta.GetSource().ToString(), 0, operation->GetCookie(), "no_tx_write", true, "");
             auto ev = NEvents::TDataEvents::TEvWriteResult::BuildCompleted(Self->TabletID());
+            AddTableAccessStatsToTxStats(*ev->Record.MutableTxStats(), writeMeta.GetPathId().SchemeShardLocalPathId.GetRawValue(),
+                                         writeResult.GetRecordsCount(), writeResult.GetDataSize(), operation->GetModificationType(), Self->TabletID());
             Results.emplace_back(std::move(ev), writeMeta.GetSource(), operation->GetCookie());
         } else {
             auto& info = Self->OperationsManager->GetLockVerified(operation->GetLockId());
@@ -84,7 +86,7 @@ bool TTxBlobsWritingFinished::DoExecute(TTransactionContext& txc, const TActorCo
             LWPROBE(EvWriteResult, Self->TabletID(), writeMeta.GetSource().ToString(), 0, operation->GetCookie(), "tx_write", true, "");
             auto ev = NEvents::TDataEvents::TEvWriteResult::BuildCompleted(Self->TabletID(), operation->GetLockId(), lock);
             AddTableAccessStatsToTxStats(*ev->Record.MutableTxStats(), writeMeta.GetPathId().SchemeShardLocalPathId.GetRawValue(),
-                                         writeResult.GetRecordsCount(), writeResult.GetDataSize(), operation->GetModificationType());
+                                         writeResult.GetRecordsCount(), writeResult.GetDataSize(), operation->GetModificationType(), Self->TabletID());
             Results.emplace_back(std::move(ev), writeMeta.GetSource(), operation->GetCookie());
         }
     }