pq: add logbroker federation support (#16293)

Author: yumkam

ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.cpp

@@ -8,6 +8,7 @@
 
 #include <ydb/library/yql/providers/pq/proto/dq_io.pb.h>
 #include <ydb/library/yql/providers/pq/proto/dq_task_params.pb.h>
+#include <ydb/library/yql/providers/pq/provider/yql_pq_gateway.h>
 
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/driver/driver.h>
 
@@ -31,10 +32,13 @@ std::pair<IDqComputeActorAsyncInput*, NActors::IActor*> CreateDqPqRdReadActor(
     ui64 taskId,
     const THashMap<TString, TString>& secureParams,
     const THashMap<TString, TString>& taskParams,
+    NYdb::TDriver driver,
+    ISecuredServiceAccountCredentialsFactory::TPtr credentialsFactory,
     const NActors::TActorId& computeActorId,
     const NActors::TActorId& localRowDispatcherActorId,
     const NKikimr::NMiniKQL::THolderFactory& holderFactory,
     const ::NMonitoring::TDynamicCounterPtr& counters,
-    i64 bufferSize = PQRdReadDefaultFreeSpace);
+    i64 bufferSize,
+    const IPqGateway::TPtr& pqGateway);
 
 } // namespace NYql::NDq

ydb/library/yql/providers/pq/async_io/dq_pq_rd_read_actor.h

@@ -76,7 +76,7 @@ void TDqPqReadActorBase::LoadState(const TSourceState& state) {
     TStringStream str;
     str << "SessionId: " << GetSessionId() << " Restoring offset: ";
     for (const auto& [key, value] : PartitionToOffset) {
-        str << "{" << key.first << "," << key.second << "," << value << "},";
+        str << "{" << key << "," << value << "},";
     }
     SRC_LOG_D(str.Str());
     StartingMessageTimestamp = minStartingMessageTs;
@@ -90,4 +90,4 @@ ui64 TDqPqReadActorBase::GetInputIndex() const {
 
 const NYql::NDq::TDqAsyncStats& TDqPqReadActorBase::GetIngressStats() const {
     return IngressStats;
-}
+}

ydb/library/yql/providers/pq/async_io/dq_pq_read_actor.cpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h>
+#include <ydb/library/yql/providers/pq/common/pq_partition_key.h>
 #include <ydb/library/yql/providers/pq/proto/dq_io.pb.h>
 #include <ydb/library/yql/providers/pq/proto/dq_task_params.pb.h>
 
@@ -9,16 +10,16 @@ namespace NYql::NDq::NInternal {
 class TDqPqReadActorBase : public IDqComputeActorAsyncInput {
 
 public:
-    using TPartitionKey = std::pair<TString, ui64>; // Cluster, partition id.
+    using TPartitionKey = ::NPq::TPartitionKey;
 
     const ui64 InputIndex;
     THashMap<TPartitionKey, ui64> PartitionToOffset; // {cluster, partition} -> offset of next event.
     const TTxId TxId;
-    const NPq::NProto::TDqPqTopicSource SourceParams;
+    NPq::NProto::TDqPqTopicSource SourceParams;
     TDqAsyncStats IngressStats;
     TInstant StartingMessageTimestamp;
     TString LogPrefix;
-    const NPq::NProto::TDqReadTaskParams ReadParams;
+    NPq::NProto::TDqReadTaskParams ReadParams;
     const NActors::TActorId ComputeActorId;
     ui64 TaskId;
 

ydb/library/yql/providers/pq/async_io/dq_pq_read_actor_base.cpp

@@ -23,6 +23,7 @@ PEERDIR(
     yql/essentials/public/types
     yql/essentials/utils/log
     ydb/public/sdk/cpp/adapters/issue
+    ydb/public/sdk/cpp/src/client/federated_topic
     ydb/public/sdk/cpp/src/client/driver
     ydb/public/sdk/cpp/src/client/topic
     ydb/public/sdk/cpp/src/client/types/credentials

ydb/library/yql/providers/pq/async_io/dq_pq_read_actor_base.h

@@ -0,0 +1,3 @@
+#include "pq_partition_key.h"
+namespace NPq {
+} // namespace NPq

ydb/library/yql/providers/pq/async_io/ya.make

@@ -0,0 +1,29 @@
+#pragma once
+#include <util/generic/hash.h>
+#include <util/generic/string.h>
+#include <util/stream/output.h>
+namespace NPq {
+    struct TPartitionKey {
+        TString Cluster;
+        ui64 PartitionId;
+        bool operator==(const TPartitionKey& other) const = default;
+        ui64 Hash() const {
+            return CombineHashes<ui64>(
+                    std::hash<TString>{} (Cluster),
+                    std::hash<ui64>{} (PartitionId)
+                    );
+        }
+    };
+}
+
+template<>
+struct THash<NPq::TPartitionKey> {
+    ui64 operator() (const NPq::TPartitionKey& x) const {
+        return x.Hash();
+    }
+};
+
+template <>
+inline void Out<NPq::TPartitionKey>(IOutputStream& stream, TTypeTraits<NPq::TPartitionKey>::TFuncParam& t) {
+    stream << t.PartitionId << '@' << t.Cluster;
+}

ydb/library/yql/providers/pq/common/pq_partition_key.cpp

@@ -2,6 +2,7 @@ LIBRARY()
 
 SRCS(
     pq_meta_fields.cpp
+    pq_partition_key.cpp
     yql_names.cpp
 )
 

ydb/library/yql/providers/pq/common/pq_partition_key.h

@@ -5,6 +5,7 @@
 namespace NYql {
 
 constexpr TStringBuf PartitionsCountProp = "PartitionsCount";
+constexpr TStringBuf FederatedClustersProp = "FederatedClusters";
 constexpr TStringBuf ConsumerSetting = "Consumer";
 constexpr TStringBuf EndpointSetting = "Endpoint";
 constexpr TStringBuf SharedReading = "SharedReading";