Merge branch 'main' into splitMerge_rescue

Author: nshestakov

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,18 @@
+### Changelog entry <!-- a user-readable short description of changes introduced in this PR -->
+
+...
+
+### Changelog category <!-- remove all except one -->
+
+* New feature
+* Experimental feature
+* Improvement
+* Performance improvement
+* Bugfix 
+* Backward incompatible change
+* Documentation (changelog entry is not required)
+* Not for changelog (changelog entry is not required)
+
+### Additional information
+
+...

.github/config/muted_ya.txt

@@ -1,3 +1,4 @@
+ydb/core/blobstorage/dsproxy/ut TBlobStorageProxySequenceTest.TestBlock42PutWithChangingSlowDisk
 ydb/core/blobstorage/pdisk/ut TSectorMap.*
 ydb/core/blobstorage/ut_blobstorage CostMetricsGetBlock4Plus2.TestGet4Plus2BlockRequests10000Inflight1BlobSize1000
 ydb/core/blobstorage/ut_blobstorage Defragmentation.DoesItWork
@@ -6,30 +7,43 @@ 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/scheme KqpOlapScheme.TenThousandColumns
 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/core/tx/schemeshard/ut_split_merge TSchemeShardSplitBySizeTest.Merge1KShards
 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
+ydb/services/persqueue_v1/ut [3/10]*
+ydb/services/ydb/sdk_sessions_pool_ut YdbSdkSessionsPool.StressTestSync*
 ydb/services/ydb/table_split_ut YdbTableSplit.SplitByLoadWithReadsMultipleSplitsWithData
 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/sqs/common test_queue_counters.py.TestSqsGettingCounters.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

ROADMAP.md

@@ -23,7 +23,7 @@ We use the following symbols as abbreviations:
 1. ㉔ Support for **temporary tables**
 1. ㉔ Support for **VIEW** SQL clause
 1. ㉔ **Data Spilling** in case there is issufient amount of RAM
-1. ㉔ **TPC-H, TPC-H for 100TB** dataset
+1. ㉔ **TPC-H, TPC-DS for 100TB** dataset
 1. ✅ ㉓ Support for **Snapshot Readonly** transactions mode
 1. 🚧 ㉓ **Better resource management** for KQP Resource Manager (share information about nodes resources, avoid OOMs)
 1. ✅ ㉓ Switch to **New Engine** for OLTP queries

ydb/core/base/logoblob.cpp

@@ -1,35 +1,26 @@
 #include "logoblob.h"
 #include <ydb/core/protos/base.pb.h>
-#include <util/string/printf.h>
 
 namespace NKikimr {
 
 TString TLogoBlobID::ToString() const {
-    return Sprintf(
-        "[%" PRIu64 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 "]",
-        TabletID(),
-        Generation(),
-        Step(),
-        Channel(),
-        Cookie(),
-        BlobSize(),
-        PartId()).data();
+    TString str;
+    str.reserve(64);
+    TStringOutput outStr(str);
+    Out(outStr);
+    return str;
 }
 
 void TLogoBlobID::Out(IOutputStream &o) const {
-    char buf[240];
-    snprintf(buf, sizeof(buf),
-        "[%" PRIu64 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 ":%" PRIu32 "]",
-        TabletID(),
-        Generation(),
-        Step(),
-        Channel(),
-        Cookie(),
-        BlobSize(),
-        PartId()
-    );
-
-    o << buf;
+    o << '['
+        << TabletID() << ':'
+        << Generation() << ':'
+        << Step() << ':'
+        << Channel() << ':'
+        << Cookie() << ':'
+        << BlobSize() << ':'
+        << PartId()
+        << ']' ;
 }
 
 void TLogoBlobID::Out(IOutputStream &o, const TVector<TLogoBlobID> &vec) {

ydb/core/base/traceid.cpp

@@ -20,16 +20,15 @@ TTraceID TTraceID::GenerateNew() {
 }
 
 TString TTraceID::ToString() const {
-    TString result;
-    TStringOutput out(result);
-    Out(out);
-    return result;
+    TString str;
+    str.reserve(128);
+    TStringOutput outStr(str);
+    Out(outStr);
+    return str;
 }
 
 void TTraceID::Out(IOutputStream &o) const {
-    char buf[240];
-    snprintf(buf, sizeof(buf), "[ID:%" PRIu64 ", Created: %s]", RandomID, TInstant::MicroSeconds(CreationTime).ToRfc822StringLocal().data());
-    o << buf;
+    o << "[ID: " << RandomID << ", " << "Created: " << TInstant::MicroSeconds(CreationTime).ToRfc822StringLocal() << "]";
 }
 
 bool TTraceID::operator<(const TTraceID &x) const {

ydb/core/formats/arrow/converter.cpp

@@ -1,9 +1,11 @@
 #include "converter.h"
 #include "switch_type.h"
 
+#include <ydb/library/binary_json/read.h>
 #include <ydb/library/binary_json/write.h>
 #include <ydb/library/dynumber/dynumber.h>
 
+#include <util/generic/set.h>
 #include <util/memory/pool.h>
 #include <library/cpp/containers/stack_vector/stack_vec.h>
 
@@ -47,16 +49,21 @@ static bool ConvertData(TCell& cell, const NScheme::TTypeInfo& colType, TMemoryP
 }
 
 static bool ConvertColumn(const NScheme::TTypeInfo colType, std::shared_ptr<arrow::Array>& column, std::shared_ptr<arrow::Field>& field) {
-    if (colType.GetTypeId() == NScheme::NTypeIds::Decimal) {
+    switch (colType.GetTypeId()) {
+    case NScheme::NTypeIds::Decimal:
         return false;
+    case NScheme::NTypeIds::JsonDocument: {
+        const static TSet<arrow::Type::type> jsonDocArrowTypes{ arrow::Type::BINARY, arrow::Type::STRING };
+        if (!jsonDocArrowTypes.contains(column->type()->id())) {
+            return false;
+        }
+        break;
     }
-
-    if ((colType.GetTypeId() == NScheme::NTypeIds::JsonDocument) &&
-        (column->type()->id() == arrow::Type::BINARY || column->type()->id() == arrow::Type::STRING))
-    {
-        ;
-    } else if (column->type()->id() != arrow::Type::BINARY) {
-        return false;
+    default:
+        if (column->type()->id() != arrow::Type::BINARY) {
+            return false;
+        }
+        break;
     }
 
     auto& binaryArray = static_cast<arrow::BinaryArray&>(*column);
@@ -81,9 +88,16 @@ static bool ConvertColumn(const NScheme::TTypeInfo colType, std::shared_ptr<arro
                     Y_ABORT_UNLESS(builder.AppendNull().ok());
                     continue;
                 }
-                const auto binaryJson = NBinaryJson::SerializeToBinaryJson(TStringBuf(value.data(), value.size()));
-                if (!binaryJson.Defined() || !builder.Append(binaryJson->Data(), binaryJson->Size()).ok()) {
-                    return false;
+                const TStringBuf valueBuf(value.data(), value.size());
+                if (NBinaryJson::IsValidBinaryJson(valueBuf)) {
+                    if (!builder.Append(value).ok()) {
+                        return false;
+                    }
+                } else {
+                    const auto binaryJson = NBinaryJson::SerializeToBinaryJson(valueBuf);
+                    if (!binaryJson.Defined() || !builder.Append(binaryJson->Data(), binaryJson->Size()).ok()) {
+                        return false;
+                    }
                 }
             }
         }

ydb/core/fq/libs/checkpointing/checkpoint_coordinator.cpp

@@ -103,6 +103,12 @@ void TCheckpointCoordinator::Handle(NYql::NDqs::TEvReadyState::TPtr& ev) {
         AllActorsSet.insert(actorId);
     }
 
+    CC_LOG_D("ActorsToTrigger count: " << ActorsToTrigger.size() << ", ActorsToNotify count: " << ActorsToNotify.size() << ", ActorsToWaitFor count: " << ActorsToWaitFor.size());
+
+    if (ActorsToTrigger.empty()) {
+        CC_LOG_D("No ingress tasks, coordinator was disabled");
+        return;
+    }
     PendingInit = std::make_unique<TPendingInitCoordinator>(AllActors.size());
 
     CC_LOG_D("Send TEvRegisterCoordinatorRequest");

ydb/core/fq/libs/checkpointing/ut/checkpoint_coordinator_ut.cpp

@@ -19,7 +19,7 @@ enum ETestGraphFlags : ui64 {
     SourceWithChannelInOneTask = 2,
 };
 
-NYql::NDqProto::TReadyState BuildTestGraph(ui64 flags = 0) {
+NYql::NDqProto::TReadyState BuildTestGraph(ui64 flags, const TString& sourceType) {
 
     NYql::NDqProto::TReadyState result;
 
@@ -29,7 +29,7 @@ NYql::NDqProto::TReadyState BuildTestGraph(ui64 flags = 0) {
     ingressOutput->AddChannels();
     if (flags & ETestGraphFlags::InputWithSource) {
         auto* source = ingress->AddInputs()->MutableSource();
-        source->SetType("PqSource");
+        source->SetType(sourceType);
     }
 
     auto* map = result.AddTask();
@@ -40,7 +40,7 @@ NYql::NDqProto::TReadyState BuildTestGraph(ui64 flags = 0) {
     mapOutput->AddChannels();
     if (flags & ETestGraphFlags::SourceWithChannelInOneTask) {
         auto* source = map->AddInputs()->MutableSource();
-        source->SetType("PqSource");
+        source->SetType(sourceType);
     }
 
     auto* egress = result.AddTask();
@@ -70,9 +70,9 @@ struct TTestBootstrap : public TTestActorRuntime {
 
     ::NMonitoring::TDynamicCounterPtr Counters = new ::NMonitoring::TDynamicCounters();
 
-    explicit TTestBootstrap(ui64 graphFlags = 0)
+    explicit TTestBootstrap(ui64 graphFlags, const TString& sourceType)
         : TTestActorRuntime(true)
-        , GraphState(BuildTestGraph(graphFlags))
+        , GraphState(BuildTestGraph(graphFlags, sourceType))
         , CoordinatorId("my-graph-id", 42)
         , CheckpointId(CoordinatorId.Generation, 1)
     {
@@ -281,8 +281,8 @@ Y_UNIT_TEST_SUITE(TCheckpointCoordinatorTests) {
     class CheckpointsTestHelper : public TTestBootstrap
     {
     public:
-        CheckpointsTestHelper(ui64 graphFlags)
-            : TTestBootstrap(graphFlags) {
+        CheckpointsTestHelper(ui64 graphFlags, const TString& sourceType)
+            : TTestBootstrap(graphFlags, sourceType) {
         }
 
         void InjectCheckpoint() {
@@ -372,22 +372,33 @@ Y_UNIT_TEST_SUITE(TCheckpointCoordinatorTests) {
     };
 
     Y_UNIT_TEST(ShouldTriggerCheckpointWithSource) {
-        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource);
+        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource, "PqSource");
         test.InjectCheckpoint();
         test.AllSavedAndCommited();
     }
 
     Y_UNIT_TEST(ShouldTriggerCheckpointWithSourcesAndWithChannel) {
-        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource | ETestGraphFlags::SourceWithChannelInOneTask);
+        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource | ETestGraphFlags::SourceWithChannelInOneTask, "PqSource");
         test.InjectCheckpoint();
         test.AllSavedAndCommited();
     }
 
     Y_UNIT_TEST(ShouldAbortPreviousCheckpointsIfNodeStateCantBeSaved) {
-        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource);
+        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource, "PqSource");
         test.InjectCheckpoint();
         test.SaveFailed();
     }
+
+    Y_UNIT_TEST(ShouldDoNothingIfNoIngressTasks) {
+        CheckpointsTestHelper test(ETestGraphFlags::InputWithSource, "S3Source");
+        bool empty = false;   
+        try {
+            test.GrabEdgeEvent<TEvCheckpointStorage::TEvRegisterCoordinatorRequest>(test.StorageProxy, TDuration::Seconds(10));
+        } catch (TEmptyEventQueueException&) {
+            empty = true;
+        }
+        UNIT_ASSERT(empty);
+    }
 }
 
 } // namespace NFq