Add colors in CLI and fix bugs

.github/config/muted_ya.txt

@@ -6,15 +6,20 @@ ydb/core/blobstorage/ut_blobstorage VDiskAssimilation.Test
 ydb/core/blobstorage/ut_blobstorage [6/10]*
 ydb/core/blobstorage/ut_blobstorage/ut_read_only_vdisk ReadOnlyVDisk.TestStorageLoad
 ydb/core/cms/ut_sentinel TSentinelTests.BSControllerCantChangeStatus
+ydb/core/persqueue/ut [31/40]*
 ydb/core/persqueue/ut TPersQueueMirrorer.TestBasicRemote
+ydb/core/quoter/ut QuoterWithKesusTest.PrefetchCoefficient
 ydb/core/kqp/ut/federated_query/generic *
 ydb/core/kqp/ut/scan KqpRequestContext.TraceIdInErrorMessage
 ydb/core/kqp/ut/service KqpQueryService.ExecuteQueryPgTableSelect
 ydb/core/tx/columnshard/ut_schema TColumnShardTestSchema.ForgetAfterFail
 ydb/core/tx/columnshard/ut_schema TColumnShardTestSchema.RebootForgetAfterFail
+ydb/core/tx/columnshard/engines/ut *
 ydb/core/tx/coordinator/ut Coordinator.RestoreTenantConfiguration
 ydb/library/yql/providers/generic/connector/tests test.py.test_select_positive_postgresql*
 ydb/library/yql/sql/pg/ut PgSqlParsingAutoparam.AutoParamValues_DifferentTypes
+ydb/library/yql/tests/sql/dq_file/part16 test.py.test[expr-as_dict_list_key-default.txt-Analyze]
+ydb/library/yql/tests/sql/dq_file/part18 test.py.test[expr-cast_type_bind-default.txt-Analyze]
 ydb/public/sdk/cpp/client/ydb_topic/ut BasicUsage.WriteRead
 ydb/services/persqueue_v1/ut TPersQueueTest.DirectRead*
 ydb/services/ydb/sdk_sessions_pool_ut YdbSdkSessionsPool.StressTestSync10
@@ -23,13 +28,17 @@ ydb/services/ydb/ut YdbOlapStore.LogPagingAfter-NotNull
 ydb/tests/fq/s3 *
 ydb/tests/fq/yds test_metrics_cleanup.py.TestCleanup.test_cleanup[v1]
 ydb/tests/functional/audit *
+ydb/tests/functional/blobstorage test_replication.py.TestReplicationAfterNodesRestart.test_replication[mirror-3-dc]
 ydb/tests/functional/clickbench test.py.test_plans*
+ydb/tests/functional/clickbench test.py.test_run_determentistic[column]
 ydb/tests/functional/kqp/kqp_query_session KqpQuerySession.NoLocalAttach
 ydb/tests/functional/postgresql test_postgres.py.TestPostgresSuite.test_postgres_suite*
 ydb/tests/functional/restarts test_restarts.py.*
 ydb/tests/functional/sqs/cloud test_yandex_cloud_mode.py.TestSqsYandexCloudMode.test_dlq_mechanics_in_cloud*
+ydb/tests/functional/sqs/cloud test_yandex_cloud_queue_counters.py.TestYmqQueueCounters.test_purge_queue_counters
 ydb/tests/functional/tenants test_dynamic_tenants.py.*
 ydb/tests/functional/tenants test_storage_config.py.TestStorageConfig.*
 ydb/tests/functional/tenants test_tenants.py.*
 ydb/tests/functional/ydb_cli test_ydb_scripting.py.TestScriptingServiceHelp.test_help
 ydb/tests/functional/ydb_cli test_ydb_scripting.py.TestScriptingServiceHelp.test_help_ex
+ydb/tests/tools/pq_read/test test_timeout.py.TestTimeout.test_timeout

ydb/core/keyvalue/keyvalue_ut_trace.cpp

@@ -0,0 +1,154 @@
+#include <ydb/core/keyvalue/keyvalue.h>
+#include <ydb/core/keyvalue/keyvalue_events.h>
+#include <ydb/core/keyvalue/protos/events.pb.h>
+#include <ydb/core/testlib/basics/helpers.h>
+#include <ydb/core/testlib/basics/runtime.h>
+#include <ydb/core/testlib/tablet_helpers.h>
+#include <ydb/library/actors/wilson/test_util/fake_wilson_uploader.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+using namespace NActors;
+using namespace NKikimr;
+
+struct TTestEnvironment {
+    THolder<TTestBasicRuntime> Runtime;
+    const ui32 NodeCount;
+    TActorId Edge;
+    const ui64 TabletId = MakeTabletID(0, 0, 1);
+    const TTabletTypes::EType TabletType = TTabletTypes::KeyValue;
+    NWilson::TFakeWilsonUploader* WilsonUploader = nullptr;
+
+    TTestEnvironment(ui32 nodeCount): NodeCount(nodeCount) {
+    }
+
+    void Prepare() {
+        SetupRuntime();
+        InitializeRuntime();
+
+        Edge = Runtime->AllocateEdgeActor();
+        CreateTestBootstrapper(*Runtime,
+            CreateTestTabletInfo(TabletId, TabletType, TErasureType::ErasureNone),
+            &CreateKeyValueFlat);
+        SetupFakeWilson();
+
+        TDispatchOptions options;
+        options.FinalEvents.push_back(TDispatchOptions::TFinalEventCondition(TEvTablet::EvBoot));
+        Runtime->DispatchEvents(options);
+    }
+
+    void InitializeRuntime() {
+        TAppPrepare app;
+        app.AddDomain(TDomainsInfo::TDomain::ConstructEmptyDomain("dc-1").Release());
+        SetupTabletServices(*Runtime, &app);
+    }
+
+    void SetupRuntime() {
+        Runtime = MakeHolder<TTestBasicRuntime>(NodeCount, 1u);
+
+        for (ui32 i = 0; i < NodeCount; ++i) {
+            SetupStateStorage(*Runtime, i, 0, true);
+            SetupTabletResolver(*Runtime, i);
+        }
+    }
+
+    void SetupFakeWilson() {
+        WilsonUploader = new NWilson::TFakeWilsonUploader;
+        auto actorId = Runtime->Register(WilsonUploader);
+        Runtime->RegisterService(NWilson::MakeWilsonUploaderId(), actorId);
+    }
+
+    template<class TRequest>
+    auto DoKVRequest(THolder<TRequest> request) {
+        Runtime->SendToPipe(TabletId, Edge, request.Release(), 0, NTabletPipe::TClientConfig(), TActorId(),
+                0, NWilson::TTraceId::NewTraceId(15, 4095));
+        TAutoPtr<IEventHandle> handle;
+        auto response = Runtime->GrabEdgeEventRethrow<typename TRequest::TResponse>(handle);
+        UNIT_ASSERT(response);
+        auto& record = response->Record;
+        UNIT_ASSERT_EQUAL(record.status(), NKikimrKeyValue::Statuses::RSTATUS_OK);
+
+        return std::move(record);
+    }
+};
+
+THolder<TEvKeyValue::TEvExecuteTransaction> CreateWrite(TString key, TString value) {
+    auto request = MakeHolder<TEvKeyValue::TEvExecuteTransaction>();
+    auto write = request->Record.add_commands()->mutable_write();
+    write->set_key(std::move(key));
+    write->set_value(std::move(value));
+    return request;
+}
+
+THolder<TEvKeyValue::TEvRead> CreateRead(TString key) {
+    auto request = MakeHolder<TEvKeyValue::TEvRead>();
+    auto& record = request->Record;
+    record.set_key(std::move(key));
+    record.set_offset(0);
+    record.set_size(0);
+    record.set_limit_bytes(0);
+    return request;
+}
+
+void TestOneWrite(TString value, TString expectedTrace) {
+    TTestEnvironment env(8);
+    env.Prepare();
+
+    env.DoKVRequest(CreateWrite("key", std::move(value)));
+
+    UNIT_ASSERT(env.WilsonUploader->BuildTraceTrees());
+    UNIT_ASSERT_EQUAL(env.WilsonUploader->Traces.size(), 1);
+    auto& trace = env.WilsonUploader->Traces.begin()->second;
+
+    UNIT_ASSERT_EQUAL(trace.ToString(), expectedTrace);
+}
+
+void TestOneRead(TString value, TString expectedTrace) {
+    TTestEnvironment env(8);
+    env.Prepare();
+
+    env.DoKVRequest(CreateWrite("key", value));
+    env.WilsonUploader->Clear();
+
+    auto response = env.DoKVRequest(CreateRead("key"));
+    UNIT_ASSERT_EQUAL(response.value(), value);
+
+    UNIT_ASSERT(env.WilsonUploader->BuildTraceTrees());
+    UNIT_ASSERT_EQUAL(env.WilsonUploader->Traces.size(), 1);
+    auto& trace = env.WilsonUploader->Traces.begin()->second;
+
+    UNIT_ASSERT_EQUAL(trace.ToString(), expectedTrace);
+}
+
+Y_UNIT_TEST_SUITE(TKeyValueTracingTest) {
+    const TString SmallValue = "value";
+    const TString HugeValue = TString(1 << 20, 'v');
+
+Y_UNIT_TEST(WriteSmall) {
+    TString canon = "(KeyValue.Intermediate -> [(KeyValue.StorageRequest -> [(DSProxy.Put -> [(Backpressure.InFlight "
+        "-> [(VDisk.Log.Put)])])]) , (Tablet.Transaction -> [(Tablet.Transaction.Execute) , (Tablet.WriteLog -> "
+        "[(Tablet.WriteLog.LogEntry -> [(DSProxy.Put -> [(Backpressure.InFlight -> [(VDisk.Log.Put)])])])])])])";
+    TestOneWrite(SmallValue, std::move(canon));
+}
+
+Y_UNIT_TEST(WriteHuge) {
+    TString canon = "(KeyValue.Intermediate -> [(KeyValue.StorageRequest -> [(DSProxy.Put -> [(Backpressure.InFlight "
+        "-> [(VDisk.HugeBlobKeeper.Write -> [(VDisk.Log.PutHuge)])])])]) , (Tablet.Transaction -> "
+        "[(Tablet.Transaction.Execute) , (Tablet.WriteLog -> [(Tablet.WriteLog.LogEntry -> [(DSProxy.Put -> "
+        "[(Backpressure.InFlight -> [(VDisk.Log.Put)])])])])])])";
+    TestOneWrite(HugeValue, std::move(canon));
+}
+
+Y_UNIT_TEST(ReadSmall) {
+    TString canon = "(KeyValue.Intermediate -> [(KeyValue.StorageReadRequest -> [(DSProxy.Get -> [(Backpressure.InFlight -> "
+        "[(VDisk.LevelIndexExtremeQueryViaBatcherMergeData)])])])])";
+    TestOneRead(SmallValue, std::move(canon));
+}
+
+Y_UNIT_TEST(ReadHuge) {
+    TString canon = "(KeyValue.Intermediate -> [(KeyValue.StorageReadRequest -> [(DSProxy.Get -> [(Backpressure.InFlight -> "
+        "[(VDisk.LevelIndexExtremeQueryViaBatcherMergeData -> [(VDisk.Query.ReadBatcher)])])])])])";
+    TestOneRead(HugeValue, std::move(canon));
+}
+
+}

ydb/core/keyvalue/ut_trace/ya.make

@@ -0,0 +1,26 @@
+UNITTEST_FOR(ydb/core/keyvalue)
+
+FORK_SUBTESTS()
+
+SPLIT_FACTOR(5)
+
+IF (SANITIZER_TYPE == "thread" OR WITH_VALGRIND)
+    TIMEOUT(1800)
+    SIZE(LARGE)
+    TAG(ya:fat)
+ELSE()
+    TIMEOUT(600)
+    SIZE(MEDIUM)
+ENDIF()
+
+PEERDIR(
+    ydb/core/testlib/default
+)
+
+SRCS(
+    keyvalue_ut_trace.cpp
+)
+
+REQUIREMENTS(ram:16)
+
+END()

ydb/core/keyvalue/ya.make

@@ -62,4 +62,5 @@ RECURSE(
 
 RECURSE_FOR_TESTS(
     ut
+    ut_trace
 )

ydb/core/kqp/executer_actor/kqp_executer_stats.cpp

@@ -330,7 +330,7 @@ bool CollectProfileStats(Ydb::Table::QueryStatsCollection::Mode statsMode) {
 void TQueryExecutionStats::FillStageDurationUs(NYql::NDqProto::TDqStageStats& stats) {
     if (stats.HasStartTimeMs() && stats.HasFinishTimeMs()) {
         auto startTimeMs = stats.GetStartTimeMs().GetMin();
-        auto finishTimeMs = stats.GetFinishTimeMs().GetMin();
+        auto finishTimeMs = stats.GetFinishTimeMs().GetMax();
         if (startTimeMs && finishTimeMs > startTimeMs) {
             stats.SetStageDurationUs((finishTimeMs - startTimeMs) * 1'000);
         }

ydb/core/kqp/ut/common/columnshard.cpp

@@ -9,7 +9,6 @@ namespace NKqp {
     TTestHelper::TTestHelper(const TKikimrSettings& settings)
         : Kikimr(settings)
         , TableClient(Kikimr.GetTableClient())
-        , LongTxClient(Kikimr.GetDriver())
         , Session(TableClient.CreateSession().GetValueSync().GetSession())
     {}
 
@@ -31,26 +30,6 @@ namespace NKqp {
         UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
     }
 
-    void TTestHelper::InsertData(const TColumnTable& table, TTestHelper::TUpdatesBuilder& updates, const std::function<void()> onBeforeCommit /*= {}*/, const EStatus opStatus /*= EStatus::SUCCESS*/) {
-        NLongTx::TLongTxBeginResult resBeginTx = LongTxClient.BeginWriteTx().GetValueSync();
-        UNIT_ASSERT_VALUES_EQUAL_C(resBeginTx.Status().GetStatus(), EStatus::SUCCESS, resBeginTx.Status().GetIssues().ToString());
-
-        auto txId = resBeginTx.GetResult().tx_id();
-        auto batch = updates.BuildArrow();
-        TString data = NArrow::NSerialization::TFullDataSerializer(arrow::ipc::IpcWriteOptions::Defaults()).Serialize(batch);
-
-        NLongTx::TLongTxWriteResult resWrite =
-            LongTxClient.Write(txId, table.GetName(), txId, data, Ydb::LongTx::Data::APACHE_ARROW).GetValueSync();
-        UNIT_ASSERT_VALUES_EQUAL_C(resWrite.Status().GetStatus(), opStatus, resWrite.Status().GetIssues().ToString());
-
-        if (onBeforeCommit) {
-            onBeforeCommit();
-        }
-
-        NLongTx::TLongTxCommitResult resCommitTx = LongTxClient.CommitTx(txId).GetValueSync();
-        UNIT_ASSERT_VALUES_EQUAL_C(resCommitTx.Status().GetStatus(), EStatus::SUCCESS, resCommitTx.Status().GetIssues().ToString());
-    }
-
     void TTestHelper::BulkUpsert(const TColumnTable& table, TTestHelper::TUpdatesBuilder& updates, const Ydb::StatusIds_StatusCode& opStatus /*= Ydb::StatusIds::SUCCESS*/) {
         Y_UNUSED(opStatus);
         NKikimr::Tests::NCS::THelper helper(Kikimr.GetTestServer());

ydb/core/kqp/ut/common/columnshard.h

@@ -3,7 +3,6 @@
 #include "kqp_ut_common.h"
 #include <ydb/library/accessor/accessor.h>
 #include <ydb/public/lib/scheme_types/scheme_type_id.h>
-#include <ydb/public/sdk/cpp/client/draft/ydb_long_tx.h>
 #include <ydb/public/sdk/cpp/client/ydb_table/table.h>
 #include <ydb/public/sdk/cpp/client/ydb_types/status_codes.h>
 #include <ydb/core/tx/columnshard/columnshard_ut_common.h>
@@ -64,7 +63,6 @@ namespace NKqp {
     private:
         TKikimrRunner Kikimr;
         NYdb::NTable::TTableClient TableClient;
-        NYdb::NLongTx::TClient LongTxClient;
         NYdb::NTable::TSession Session;
 
     public:
@@ -73,7 +71,6 @@ namespace NKqp {
         TTestActorRuntime& GetRuntime();
         NYdb::NTable::TSession& GetSession();
         void CreateTable(const TColumnTableBase& table);
-        void InsertData(const TColumnTable& table, TTestHelper::TUpdatesBuilder& updates, const std::function<void()> onBeforeCommit = {}, const NYdb::EStatus opStatus = NYdb::EStatus::SUCCESS);
         void BulkUpsert(const TColumnTable& table, TTestHelper::TUpdatesBuilder& updates, const Ydb::StatusIds_StatusCode& opStatus = Ydb::StatusIds::SUCCESS);
         void BulkUpsert(const TColumnTable& table, std::shared_ptr<arrow::RecordBatch> batch, const Ydb::StatusIds_StatusCode& opStatus = Ydb::StatusIds::SUCCESS);
         void ReadData(const TString& query, const TString& expected, const NYdb::EStatus opStatus = NYdb::EStatus::SUCCESS);

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

@@ -48,7 +48,7 @@ Y_UNIT_TEST_SUITE(KqpOlapStats) {
                 tableInserter.AddRow().Add(i).Add("test_res_" + std::to_string(i)).AddNull();
             }
 
-            testHelper.InsertData(testTable, tableInserter);
+            testHelper.BulkUpsert(testTable, tableInserter);
         }
 
         Sleep(TDuration::Seconds(1));
@@ -88,7 +88,7 @@ Y_UNIT_TEST_SUITE(KqpOlapStats) {
             for (size_t i = 0; i < inserted_rows; i++) {
                 tableInserter.AddRow().Add(i).Add("test_res_" + std::to_string(i)).AddNull();
             }
-            testHelper.InsertData(testTable, tableInserter);
+            testHelper.BulkUpsert(testTable, tableInserter);
         }
 
         Sleep(TDuration::Seconds(1));
@@ -135,7 +135,7 @@ Y_UNIT_TEST_SUITE(KqpOlapStats) {
                     .Add("test_res_" + std::to_string(i + t * tables_in_store))
                     .AddNull();
             }
-            testHelper.InsertData(testTable, tableInserter);
+            testHelper.BulkUpsert(testTable, tableInserter);
         }
 
         Sleep(TDuration::Seconds(20));
@@ -155,4 +155,4 @@ Y_UNIT_TEST_SUITE(KqpOlapStats) {
 }
 
 }   // namespace NKqp
-}   // namespace NKikimr
+}   // namespace NKikimr