24-3: parse topic's partitioning once and more efficiently, better ha…

…ndling of connection loss (ydb-platform#10028)

ydb/core/change_exchange/change_sender_common_ops.h

@@ -413,8 +413,10 @@ class TBaseChangeSender {
     }
 
     TActorId GetChangeServer() const { return ChangeServer; }
-    void CreateSenders(const TVector<ui64>& partitionIds, bool partitioningChanged = true) {
-        if (partitioningChanged) {
+
+private:
+    void CreateSendersImpl(const TVector<ui64>& partitionIds) {
+        if (partitionIds) {
             CreateMissingSenders(partitionIds);
         } else {
             RecreateSenders(GonePartitions);
@@ -427,6 +429,16 @@ class TBaseChangeSender {
         }
     }
 
+protected:
+    void CreateSenders(const TVector<ui64>& partitionIds) {
+        Y_ABORT_UNLESS(partitionIds);
+        CreateSendersImpl(partitionIds);
+    }
+
+    void CreateSenders() {
+        CreateSendersImpl({});
+    }
+
     void KillSenders() {
         for (const auto& [_, sender] : std::exchange(Senders, {})) {
             if (sender.ActorId) {

ydb/core/change_exchange/util.cpp

@@ -0,0 +1,15 @@
+#include "util.h"
+
+namespace NKikimr::NChangeExchange {
+
+TVector<ui64> MakePartitionIds(const TVector<TKeyDesc::TPartitionInfo>& partitions) {
+    TVector<ui64> result(::Reserve(partitions.size()));
+
+    for (const auto& partition : partitions) {
+        result.push_back(partition.ShardId);
+    }
+
+    return result;
+}
+
+}

ydb/core/change_exchange/util.h

@@ -0,0 +1,9 @@
+#pragma once
+
+#include <ydb/core/scheme/scheme_tabledefs.h>
+
+namespace NKikimr::NChangeExchange {
+
+TVector<ui64> MakePartitionIds(const TVector<TKeyDesc::TPartitionInfo>& partitions);
+
+}

ydb/core/change_exchange/ya.make

@@ -4,6 +4,7 @@ SRCS(
     change_exchange.cpp
     change_record.cpp
     change_sender_monitoring.cpp
+    util.cpp
 )
 
 GENERATE_ENUM_SERIALIZATION(change_record.h)

ydb/core/tx/datashard/change_sender_async_index.cpp

@@ -6,6 +6,7 @@
 #include <ydb/core/base/tablet_pipecache.h>
 #include <ydb/core/change_exchange/change_sender_common_ops.h>
 #include <ydb/core/change_exchange/change_sender_monitoring.h>
+#include <ydb/core/change_exchange/util.h>
 #include <ydb/core/tablet_flat/flat_row_eggs.h>
 #include <ydb/core/tx/scheme_cache/helpers.h>
 #include <ydb/core/tx/scheme_cache/scheme_cache.h>
@@ -435,16 +436,6 @@ class TAsyncIndexChangeSenderMain
         return Check(&TSchemeCacheHelpers::CheckEntryKind<T>, &TThis::LogWarnAndRetry, entry, expected);
     }
 
-    static TVector<ui64> MakePartitionIds(const TVector<TKeyDesc::TPartitionInfo>& partitions) {
-        TVector<ui64> result(Reserve(partitions.size()));
-
-        for (const auto& partition : partitions) {
-            result.push_back(partition.ShardId); // partition = shard
-        }
-
-        return result;
-    }
-
     /// ResolveUserTable
 
     void ResolveUserTable() {
@@ -611,6 +602,11 @@ class TAsyncIndexChangeSenderMain
             return;
         }
 
+        if (IndexTableVersion && IndexTableVersion == entry.Self->Info.GetVersion().GetGeneralVersion()) {
+            CreateSenders();
+            return Become(&TThis::StateMain);
+        }
+
         TagMap.clear();
         TVector<NScheme::TTypeInfo> keyColumnTypes;
 
@@ -692,11 +688,9 @@ class TAsyncIndexChangeSenderMain
             return Retry();
         }
 
-        const bool versionChanged = !IndexTableVersion || IndexTableVersion != entry.GeneralVersion;
         IndexTableVersion = entry.GeneralVersion;
-
         KeyDesc = std::move(entry.KeyDescription);
-        CreateSenders(MakePartitionIds(KeyDesc->GetPartitions()), versionChanged);
+        CreateSenders(NChangeExchange::MakePartitionIds(KeyDesc->GetPartitions()));
 
         Become(&TThis::StateMain);
     }

ydb/core/tx/datashard/change_sender_cdc_stream.cpp

@@ -6,7 +6,7 @@
 
 #include <ydb/core/change_exchange/change_sender_common_ops.h>
 #include <ydb/core/change_exchange/change_sender_monitoring.h>
-#include <ydb/core/persqueue/partition_key_range/partition_key_range.h>
+#include <ydb/core/change_exchange/util.h>
 #include <ydb/core/persqueue/writer/source_id_encoding.h>
 #include <ydb/core/persqueue/writer/writer.h>
 #include <ydb/core/tx/scheme_cache/helpers.h>
@@ -300,45 +300,6 @@ class TCdcChangeSenderMain
     , public NChangeExchange::ISenderFactory
     , private NSchemeCache::TSchemeCacheHelpers
 {
-    struct TPQPartitionInfo {
-        ui32 PartitionId;
-        ui64 ShardId;
-        TPartitionKeyRange KeyRange;
-
-        struct TLess {
-            TConstArrayRef<NScheme::TTypeInfo> Schema;
-
-            TLess(const TVector<NScheme::TTypeInfo>& schema)
-                : Schema(schema)
-            {
-            }
-
-            bool operator()(const TPQPartitionInfo& lhs, const TPQPartitionInfo& rhs) const {
-                Y_ABORT_UNLESS(lhs.KeyRange.ToBound || rhs.KeyRange.ToBound);
-
-                if (!lhs.KeyRange.ToBound) {
-                    return false;
-                }
-
-                if (!rhs.KeyRange.ToBound) {
-                    return true;
-                }
-
-                Y_ABORT_UNLESS(lhs.KeyRange.ToBound && rhs.KeyRange.ToBound);
-
-                const int compares = CompareTypedCellVectors(
-                    lhs.KeyRange.ToBound->GetCells().data(),
-                    rhs.KeyRange.ToBound->GetCells().data(),
-                    Schema.data(), Schema.size()
-                );
-
-                return (compares < 0);
-            }
-
-        }; // TLess
-
-    }; // TPQPartitionInfo
-
     TStringBuf GetLogPrefix() const {
         if (!LogPrefix) {
             LogPrefix = TStringBuilder()
@@ -430,16 +391,6 @@ class TCdcChangeSenderMain
         return false;
     }
 
-    static TVector<ui64> MakePartitionIds(const TVector<NKikimr::TKeyDesc::TPartitionInfo>& partitions) {
-        TVector<ui64> result(Reserve(partitions.size()));
-
-        for (const auto& partition : partitions) {
-            result.push_back(partition.ShardId);
-        }
-
-        return result;
-    }
-
     /// ResolveCdcStream
 
     void ResolveCdcStream() {
@@ -561,77 +512,27 @@ class TCdcChangeSenderMain
             return;
         }
 
-        const auto& pqDesc = entry.PQGroupInfo->Description;
-        const auto& pqConfig = pqDesc.GetPQTabletConfig();
-
-        TVector<NScheme::TTypeInfo> schema;
-        PartitionToShard.clear();
-
-        schema.reserve(pqConfig.PartitionKeySchemaSize());
-        for (const auto& keySchema : pqConfig.GetPartitionKeySchema()) {
-            // TODO: support pg types
-            schema.push_back(NScheme::TTypeInfo(keySchema.GetTypeId()));
+        const auto topicVersion = entry.Self->Info.GetVersion().GetGeneralVersion();
+        if (TopicVersion && TopicVersion == topicVersion) {
+            CreateSenders();
+            return Become(&TThis::StateMain);
         }
 
-        TSet<TPQPartitionInfo, TPQPartitionInfo::TLess> partitions(schema);
-        THashSet<ui64> shards;
-
-        for (const auto& partition : pqDesc.GetPartitions()) {
-            const auto partitionId = partition.GetPartitionId();
-            const auto shardId = partition.GetTabletId();
-
-            PartitionToShard.emplace(partitionId, shardId);
-
-            auto keyRange = TPartitionKeyRange::Parse(partition.GetKeyRange());
-            Y_ABORT_UNLESS(!keyRange.FromBound || keyRange.FromBound->GetCells().size() == schema.size());
-            Y_ABORT_UNLESS(!keyRange.ToBound || keyRange.ToBound->GetCells().size() == schema.size());
-
-            partitions.insert({partitionId, shardId, std::move(keyRange)});
-            shards.insert(shardId);
-        }
-
-        // used to validate
-        bool isFirst = true;
-        const TPQPartitionInfo* prev = nullptr;
-
-        TVector<NKikimr::TKeyDesc::TPartitionInfo> partitioning;
-        partitioning.reserve(partitions.size());
-        for (const auto& cur : partitions) {
-            if (isFirst) {
-                isFirst = false;
-                Y_ABORT_UNLESS(!cur.KeyRange.FromBound.Defined());
-            } else {
-                Y_ABORT_UNLESS(cur.KeyRange.FromBound.Defined());
-                Y_ABORT_UNLESS(prev);
-                Y_ABORT_UNLESS(prev->KeyRange.ToBound.Defined());
-                // TODO: compare cells
-            }
-
-            auto& part = partitioning.emplace_back(cur.PartitionId); // TODO: double-check that it is right partitioning
-
-            if (cur.KeyRange.ToBound) {
-                part.Range = NKikimr::TKeyDesc::TPartitionRangeInfo{
-                    .EndKeyPrefix = *cur.KeyRange.ToBound,
-                };
-            } else {
-                part.Range = NKikimr::TKeyDesc::TPartitionRangeInfo{};
-            }
+        TopicVersion = topicVersion;
 
-            prev = &cur;
-        }
+        const auto& pqDesc = entry.PQGroupInfo->Description;
 
-        if (prev) {
-            Y_ABORT_UNLESS(!prev->KeyRange.ToBound.Defined());
+        PartitionToShard.clear();
+        for (const auto& partition : pqDesc.GetPartitions()) {
+            PartitionToShard.emplace(partition.GetPartitionId(), partition.GetTabletId());
         }
 
-        const auto topicVersion = entry.Self->Info.GetVersion().GetGeneralVersion();
-        const bool versionChanged = !TopicVersion || TopicVersion != topicVersion;
-        TopicVersion = topicVersion;
-
-        KeyDesc = NKikimr::TKeyDesc::CreateMiniKeyDesc(schema);
-        KeyDesc->Partitioning = std::make_shared<TVector<NKikimr::TKeyDesc::TPartitionInfo>>(std::move(partitioning));
+        Y_ABORT_UNLESS(entry.PQGroupInfo->Schema);
+        KeyDesc = NKikimr::TKeyDesc::CreateMiniKeyDesc(entry.PQGroupInfo->Schema);
+        Y_ABORT_UNLESS(entry.PQGroupInfo->Partitioning);
+        KeyDesc->Partitioning = std::make_shared<TVector<NKikimr::TKeyDesc::TPartitionInfo>>(entry.PQGroupInfo->Partitioning);
 
-        CreateSenders(MakePartitionIds(*KeyDesc->Partitioning), versionChanged);
+        CreateSenders(NChangeExchange::MakePartitionIds(*KeyDesc->Partitioning));
         Become(&TThis::StateMain);
     }
 

ydb/core/tx/datashard/ya.make

@@ -244,7 +244,6 @@ PEERDIR(
     ydb/core/formats
     ydb/core/io_formats/ydb_dump
     ydb/core/kqp/runtime
-    ydb/core/persqueue/partition_key_range
     ydb/core/persqueue/writer
     ydb/core/protos
     ydb/core/tablet

ydb/core/tx/replication/service/table_writer_impl.h

@@ -6,6 +6,7 @@
 
 #include <ydb/core/base/tablet_pipecache.h>
 #include <ydb/core/change_exchange/change_sender_common_ops.h>
+#include <ydb/core/change_exchange/util.h>
 #include <ydb/core/scheme/scheme_tabledefs.h>
 #include <ydb/core/tablet_flat/flat_row_eggs.h>
 #include <ydb/core/tx/datashard/datashard.h>
@@ -278,16 +279,6 @@ class TLocalTableWriter
         return Check(&TSchemeCacheHelpers::CheckEntryKind<T>, &TThis::LogCritAndLeave, entry, expected);
     }
 
-    static TVector<ui64> MakePartitionIds(const TVector<TKeyDesc::TPartitionInfo>& partitions) {
-        TVector<ui64> result(::Reserve(partitions.size()));
-
-        for (const auto& partition : partitions) {
-            result.push_back(partition.ShardId);
-        }
-
-        return result;
-    }
-
     void Registered(TActorSystem*, const TActorId&) override {
         this->ChangeServer = this->SelfId();
     }
@@ -348,6 +339,12 @@ class TLocalTableWriter
             return;
         }
 
+        if (TableVersion && TableVersion == entry.Self->Info.GetVersion().GetGeneralVersion()) {
+            Y_ABORT_UNLESS(Initialized);
+            Resolving = false;
+            return this->CreateSenders();
+        }
+
         auto schema = MakeIntrusive<TLightweightSchema>();
         if (entry.Self && entry.Self->Info.HasVersion()) {
             schema->Version = entry.Self->Info.GetVersion().GetTableSchemaVersion();
@@ -415,11 +412,9 @@ class TLocalTableWriter
             return LogWarnAndRetry("Empty partitions");
         }
 
-        const bool versionChanged = !TableVersion || TableVersion != entry.GeneralVersion;
         TableVersion = entry.GeneralVersion;
-
         KeyDesc = std::move(entry.KeyDescription);
-        this->CreateSenders(MakePartitionIds(KeyDesc->GetPartitions()), versionChanged);
+        this->CreateSenders(NChangeExchange::MakePartitionIds(KeyDesc->GetPartitions()));
 
         if (!Initialized) {
             this->Send(Worker, new TEvWorker::TEvHandshake());