Ydb stable 23-4-4

x-stable-origin-commit: d7fd6fddf6a14d5723ab64f6b8d512a7f239bc88

Author: Gazizonoki

library/cpp/actors/testlib/test_runtime.cpp

@@ -1202,11 +1202,21 @@ namespace NActors {
                                     isEdgeMailbox = true;
                                     TEventsList events;
                                     mbox.second->Capture(events);
+
+                                    TEventsList eventsToPush;
                                     for (auto& ev : events) {
                                         TInverseGuard<TMutex> inverseGuard(Mutex);
-                                        ObserverFunc(*this, ev);
+
+                                        for (auto& observer : ObserverFuncs) {
+                                            observer(ev);
+                                            if (!ev) break;
+                                        }
+
+                                        if (ev && ObserverFunc(*this, ev) != EEventAction::DROP) {
+                                            eventsToPush.push_back(ev);
+                                        }
                                     }
-                                    mbox.second->PushFront(events);
+                                    mbox.second->PushFront(eventsToPush);
                                 }
 
                                 if (!isEdgeMailbox) {
@@ -1232,7 +1242,17 @@ namespace NActors {
                                     EEventAction action;
                                     {
                                         TInverseGuard<TMutex> inverseGuard(Mutex);
-                                        action = ObserverFunc(*this, ev);
+
+                                        for (auto& observer : ObserverFuncs) {
+                                            observer(ev);
+                                            if (!ev) break;
+                                        }
+
+                                        if (ev) {
+                                            action = ObserverFunc(*this, ev);
+                                        } else {
+                                            action = EEventAction::DROP;
+                                        }
                                     }
 
                                     switch (action) {

library/cpp/actors/testlib/test_runtime.h

@@ -296,6 +296,52 @@ namespace NActors {
         TIntrusivePtr<NMonitoring::TDynamicCounters> GetDynamicCounters(ui32 nodeIndex = 0);
         void SetupMonitoring();
 
+        using TEventObserverCollection = std::list<std::function<void(TAutoPtr<IEventHandle>& event)>>;
+        class TEventObserverHolder {
+        public:
+            TEventObserverHolder(TEventObserverCollection& list, TEventObserverCollection::iterator&& iter)
+                : List(list)
+                , Iter(iter)
+            {
+            }
+
+            ~TEventObserverHolder()
+            {
+                Remove();
+            }
+
+            void Remove()
+            {
+                if (Iter == List.end()) {
+                    return;
+                }
+
+                List.erase(Iter);
+                Iter = List.end();
+            }
+        private:
+            TEventObserverCollection& List;
+            TEventObserverCollection::iterator Iter;
+        };
+
+        template <typename TEvType>
+        TEventObserverHolder AddObserver(std::function<void(typename TEvType::TPtr&)> observerFunc)
+        {
+            auto baseFunc = [observerFunc](TAutoPtr<IEventHandle>& event) {
+                if (event && event->GetTypeRewrite() == TEvType::EventType)
+                    observerFunc(*(reinterpret_cast<typename TEvType::TPtr*>(&event)));
+            };
+
+            auto iter = ObserverFuncs.insert(ObserverFuncs.end(), baseFunc);
+            return TEventObserverHolder(ObserverFuncs, std::move(iter));
+        }
+
+        TEventObserverHolder AddObserver(std::function<void(TAutoPtr<IEventHandle>&)> observerFunc)
+        {
+            auto iter = ObserverFuncs.insert(ObserverFuncs.end(), observerFunc);
+            return TEventObserverHolder(ObserverFuncs, std::move(iter));
+        }
+
         template<typename T>
         void AppendToLogSettings(NLog::EComponent minVal, NLog::EComponent maxVal, T func) {
             Y_VERIFY(!IsInitialized);
@@ -653,6 +699,7 @@ namespace NActors {
         TDuration DispatchTimeout;
         TDuration ReschedulingDelay;
         TEventObserver ObserverFunc;
+        TEventObserverCollection ObserverFuncs;
         TScheduledEventsSelector ScheduledEventsSelectorFunc;
         TEventFilter EventFilterFunc;
         TScheduledEventFilter ScheduledEventFilterFunc;

library/cpp/containers/disjoint_interval_tree/disjoint_interval_tree.h

@@ -132,7 +132,7 @@ class TDisjointIntervalTree {
         }
 
         TIterator completelyRemoveEnd = completelyRemoveBegin != Tree.end() ? Tree.lower_bound(end) : Tree.end();
-        if (completelyRemoveEnd != Tree.end() && completelyRemoveEnd != Tree.begin() && completelyRemoveEnd->first != end) {
+        if (completelyRemoveEnd != Tree.begin() && (completelyRemoveEnd == Tree.end() || completelyRemoveEnd->first != end)) {
             TIterator containingEnd = completelyRemoveEnd;
             --containingEnd;
             if (containingEnd->second > end) {

library/cpp/containers/disjoint_interval_tree/ut/disjoint_interval_tree_ut.cpp

@@ -232,6 +232,18 @@ Y_UNIT_TEST_SUITE(DisjointIntervalTreeTest) {
             UNIT_ASSERT_VALUES_EQUAL(tree.GetNumIntervals(), 2);
             UNIT_ASSERT_VALUES_EQUAL(tree.GetNumElements(), 8);
         }
+
+        // 12. The only one interval
+        {
+            TDisjointIntervalTree<ui64> tree;
+            tree.InsertInterval(1, 10);
+            UNIT_ASSERT_VALUES_EQUAL(tree.GetNumIntervals(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(tree.GetNumElements(), 9);
+            UNIT_ASSERT_VALUES_EQUAL(tree.EraseInterval(0, 6), 5);
+            UNIT_ASSERT_VALUES_EQUAL(tree.GetNumIntervals(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(tree.GetNumElements(), 4);
+            UNIT_ASSERT(tree.Intersects(5, 10));
+        }
     }
 
     Y_UNIT_TEST(IntersectsTest) {

ydb/core/driver_lib/version/version.cpp

@@ -28,7 +28,7 @@ TCompatibilityInfo::TCompatibilityInfo() {
         .Application = "ydb",
         .Version = TVersionConstructor{
             .Year = 23,
-            .Major = 3,
+            .Major = 4,
         }
     }.ToPB();
 
@@ -79,13 +79,13 @@ const TStored* TCompatibilityInfo::GetDefault(TComponentId componentId) const {
 // obsolete version control
 TMaybe<NActors::TInterconnectProxyCommon::TVersionInfo> VERSION = NActors::TInterconnectProxyCommon::TVersionInfo{
     // version of this binary
-    "stable-23-3",
+    "stable-23-4",
 
     // compatible versions; must include all compatible old ones, including this one; version verification occurs on both
     // peers and connection is accepted if at least one of peers accepts the version of the other peer
     {
-        "stable-23-2",
-        "stable-23-3"
+        "stable-23-3",
+        "stable-23-4"
     }
 };
 

ydb/core/persqueue/writer/writer.cpp

@@ -205,11 +205,7 @@ class TPartitionWriter: public TActorBootstrapped<TPartitionWriter>, private TRl
         auto ev = MakeRequest(PartitionId, PipeClient);
 
         auto& cmd = *ev->Record.MutablePartitionRequest()->MutableCmdGetOwnership();
-        if (Opts.UseDeduplication) {
-            cmd.SetOwner(SourceId);
-        } else {
-            cmd.SetOwner(CreateGuidAsString());
-        }
+        cmd.SetOwner(SourceId);
         cmd.SetForce(true);
 
         NTabletPipe::SendData(SelfId(), PipeClient, ev.Release());
@@ -659,7 +655,7 @@ class TPartitionWriter: public TActorBootstrapped<TPartitionWriter>, private TRl
         , TabletId(tabletId)
         , PartitionId(partitionId)
         , ExpectedGeneration(expectedGeneration)
-        , SourceId(sourceId)
+        , SourceId(opts.UseDeduplication ? sourceId : CreateGuidAsString())
         , Opts(opts)
     {
         if (Opts.MeteringMode) {

ydb/core/protos/flat_tx_scheme.proto

@@ -103,12 +103,13 @@ message TEvDescribeSchemeResult {
     optional string Reason = 2;
     optional string Path = 3;
     optional NKikimrSchemeOp.TPathDescription PathDescription = 4;
-    optional fixed64 PathOwner = 5;
+    optional fixed64 DEPRECATED_PathOwner = 5;  // replaced by PathOwnerId
     optional fixed64 PathId = 6;
 
     optional string LastExistedPrefixPath = 7;
     optional fixed64 LastExistedPrefixPathId = 8;
     optional NKikimrSchemeOp.TPathDescription LastExistedPrefixDescription = 9;
+
     optional fixed64 PathOwnerId = 10;
 }
 

ydb/core/protos/scheme_board.proto

@@ -1,4 +1,4 @@
-import "ydb/core/protos/flat_tx_scheme.proto";
+import "ydb/core/protos/base.proto";
 
 package NKikimrSchemeBoard;
 option java_package = "ru.yandex.kikimr.proto";
@@ -13,22 +13,76 @@ message TEvHandshake {
     optional uint64 Generation = 2;
 }
 
-// here and below
-// Owner is the tablet id of schemeshard witch holds the records
-// LocalPathId is a second part of TPathId
-// PathOwnerId is a first part of TPathId
+// Here and below.
+// Owner is the tablet id of schemeshard which holds the records.
+// (PathOwnerId, LocalPathId) constitute TPathId of the object.
 
+// TEvUpdate.DescribeSchemeResultSerialized is a NKikimrScheme.TEvDescribeSchemeResult
+// in the form of opaque payload.
+// Originally, that field existed as a properly typed TEvDescribeSchemeResult message.
+// However, that induce additional overhead to serialize and deserialize this message
+// when transferring over wire.
+// This performance cost is usually either negligible or imperceptible.
+// But in specific situations, particularly when rapidly updating partitioning information
+// for tables with huge number of shards, this overhead could lead to significant issues.
+// Schemeboard replicas could get overloaded and become unresponsive to further requests.
+// This is problematic, especially considering the schemeboard subsystem's critical role
+// in servicing all databases within a cluster, making it a Single Point of Failure (SPOF).
+//
+// The core realization is that the schemeboard components do not require the full content of
+// a TEvDescribeSchemeResult message to operate efficiently. Instead, only a limited set of
+// fields (path, path-id, version and info about subdomain/database) is required for processing.
+// And a whole TEvDescribeSchemeResult could be passed through as an opaque payload.
+//
+// Type change from TEvDescribeSchemeResult to (repeated) bytes without changing field number
+// is a safe move. Actual value of the field remains unchanged at the wire-format level.
+// Thus, older implementations will interpret the payload as a TEvDescribeSchemeResult message
+// and proceed with deserialization as usual. And newer implementations will recognize the data
+// as a binary blob and will deserialize it explicitly only when necessary.
+//
+// Note that the `repeated` label for the `DescribeSchemeResultSerialized` field is essential
+// to remain backward-compatible with the previous implementation. This is because even if
+// DescribeSchemeResult previously was labeled `optional` but actual value used at
+// the wire-format level was (and is) a pack of TEvDescribeSchemeResult messages.
+// Automerge of consecutive messages for the same field is a prominent feature of the protobuf.
+// Schemeshard use that feature to supply full TEvDescribeSchemeResult as a sequence of
+// partially filled TEvDescribeSchemeResult's.
+//
+// - Path
+// - PathOwnerId, LocalPathId
+// - PathDirEntryPathVersion
+// - PathSubdomainPathId
+// - PathAbandonedTenantsSchemeShards
+// are taken from the original TEvDescribeSchemeResult (one way or another).
+//
 message TEvUpdate {
     optional uint64 Owner = 1;
     optional uint64 Generation = 2;
     optional TLocalPathIdRange DeletedLocalPathIds = 3;
-    optional string Path = 4;
-    optional uint64 LocalPathId = 5;
+
+    optional string Path = 4;  // extracted from DescribeSchemeResult.Path
+    optional uint64 LocalPathId = 5;  // extracted from DescribeSchemeResult.PathId
+
     optional bool IsDeletion = 6 [default = false];
-    optional NKikimrScheme.TEvDescribeSchemeResult DescribeSchemeResult = 7;
+
+    repeated bytes DescribeSchemeResultSerialized = 7;
+
     optional bool NeedAck = 8 [default = false];
-    optional uint64 PathOwnerId = 9;
+
+    optional uint64 PathOwnerId = 9;  // extracted from DescribeSchemeResult.PathOwnerId, DescribeSchemeResult.PathDescription.Self.SchemeshardId in order of presence
+
     optional TLocalPathIdRange MigratedLocalPathIds = 10;
+
+    // Explicit values extracted from DescribeSchemeResultSerialized
+
+    //  DescribeSchemeResult.PathDescription.Self.PathVersion
+    optional uint64 PathDirEntryPathVersion = 11;
+
+    //  DescribeSchemeResult.PathDescription.DomainDescription.DomainKey
+    optional NKikimrProto.TPathID PathSubdomainPathId = 13;
+
+    //  DescribeSchemeResult.PathDescription.AbandonedTenantsSchemeShards
+    repeated uint64 PathAbandonedTenantsSchemeShards = 14;
 }
 
 message TEvUpdateAck {
@@ -65,16 +119,22 @@ message TEvUnsubscribe {
     optional uint64 LocalPathId = 3;
 }
 
+ // See comments for TEvUpdate.
 message TEvNotify {
     optional string Path = 1;
     // and/or
     optional uint64 PathOwnerId = 2;
     optional uint64 LocalPathId = 3;
     // common fields
     optional bool IsDeletion = 4 [default = false];
-    optional NKikimrScheme.TEvDescribeSchemeResult DescribeSchemeResult = 5;
-    optional uint64 Version = 6;
+
+    optional bytes DescribeSchemeResultSerialized = 5;
+
+    optional uint64 Version = 6;  // same as TEvUpdate.PathDirEntryPathVersion
     optional bool Strong = 7 [default = false];
+
+    optional NKikimrProto.TPathID PathSubdomainPathId = 8;
+    repeated uint64 PathAbandonedTenantsSchemeShards = 9;
 }
 
 message TEvNotifyAck {

ydb/core/tx/columnshard/blobs_action/counters/storage.cpp

@@ -9,11 +9,8 @@ TStorageCounters::TStorageCounters(const TString& storageId)
 }
 
 std::shared_ptr<NKikimr::NOlap::NBlobOperations::TConsumerCounters> TStorageCounters::GetConsumerCounter(const TString& consumerId) {
-    auto it = ConsumerCounters.find(consumerId);
-    if (it == ConsumerCounters.end()) {
-        it = ConsumerCounters.emplace(consumerId, std::make_shared<TConsumerCounters>(consumerId, *this)).first;
-    }
-    return it->second;
+    Y_UNUSED(ConsumerCounters);
+    return std::make_shared<TConsumerCounters>(consumerId, *this);
 }
 
 TConsumerCounters::TConsumerCounters(const TString& consumerId, const TStorageCounters& parent)

ydb/core/tx/columnshard/blobs_action/tier/storage.cpp

@@ -48,6 +48,10 @@ void TOperator::InitNewExternalOperator(const NColumnShard::NTiers::TManager* ti
     } else {
         settings.SetEndpoint("nowhere");
     }
+    if (CurrentS3Settings && CurrentS3Settings->SerializeAsString() == settings.SerializeAsString()) {
+        return;
+    }
+    CurrentS3Settings = settings;
     auto extStorageConfig = NWrappers::NExternalStorage::IExternalStorageConfig::Construct(settings);
     AFL_VERIFY(extStorageConfig);
     auto extStorageOperator = extStorageConfig->ConstructStorageOperator(false);

ydb/core/tx/columnshard/blobs_action/tier/storage.h

@@ -17,6 +17,7 @@ class TOperator: public IBlobsStorageOperator {
     TAtomicCounter CurrentOperatorIdx = 0;
     std::deque<NWrappers::NExternalStorage::IExternalStorageOperator::TPtr> ExternalStorageOperators;
     std::shared_ptr<TGCInfo> GCInfo = std::make_shared<TGCInfo>();
+    std::optional<NKikimrSchemeOp::TS3Settings> CurrentS3Settings;
     NWrappers::NExternalStorage::IExternalStorageConfig::TPtr ExternalStorageConfig;
     NWrappers::NExternalStorage::IExternalStorageOperator::TPtr ExternalStorageOperator;
 

ydb/core/tx/columnshard/engines/insert_table/rt_insertion.cpp

@@ -115,6 +115,9 @@ THashSet<NKikimr::NOlap::TWriteId> TInsertionSummary::GetDeprecatedInsertions(co
     for (auto& [writeId, data] : Inserted) {
         if (data.GetMeta().GetDirtyWriteTime() && data.GetMeta().GetDirtyWriteTime() < timeBorder) {
             toAbort.insert(writeId);
+            if (toAbort.size() == 20000) {
+                break;
+            }
         }
     }
     return toAbort;

ydb/core/tx/columnshard/engines/portions/column_record.cpp

@@ -14,10 +14,6 @@ TChunkMeta::TChunkMeta(const TColumnChunkLoadContext& context, const TIndexInfo&
     if (context.GetMetaProto().HasRawBytes()) {
         RawBytes = context.GetMetaProto().GetRawBytes();
     }
-    if (context.GetMetaProto().HasMinValue()) {
-        AFL_VERIFY(field)("field_id", context.GetAddress().GetColumnId())("field_name", indexInfo.GetColumnName(context.GetAddress().GetColumnId()));
-        Min = ConstantToScalar(context.GetMetaProto().GetMinValue(), field->type());
-    }
     if (context.GetMetaProto().HasMaxValue()) {
         AFL_VERIFY(field)("field_id", context.GetAddress().GetColumnId())("field_name", indexInfo.GetColumnName(context.GetAddress().GetColumnId()));
         Max = ConstantToScalar(context.GetMetaProto().GetMaxValue(), field->type());
@@ -37,9 +33,9 @@ NKikimrTxColumnShard::TIndexColumnMeta TChunkMeta::SerializeToProto() const {
     if (RawBytes) {
         meta.SetRawBytes(*RawBytes);
     }
-    if (HasMinMax()) {
-        ScalarToConstant(*Min, *meta.MutableMinValue());
+    if (HasMax()) {
         ScalarToConstant(*Max, *meta.MutableMaxValue());
+        ScalarToConstant(*Max, *meta.MutableMinValue());
     }
     return meta;
 }