issues (except logging)

Author: ildar-khisambeev

ydb/public/sdk/cpp/client/ydb_federated_topic/federated_topic.h

@@ -4,6 +4,8 @@
 
 #include <ydb/public/api/protos/ydb_federation_discovery.pb.h>
 
+#include <ydb/public/sdk/cpp/client/ydb_types/exceptions/exceptions.h>
+
 #include <unordered_set>
 
 namespace NYdb::NFederatedTopic {
@@ -397,9 +399,18 @@ struct TFederatedReadSessionSettings: public NTopic::TReadSessionSettings {
     //! Discards previously set ReadOriginal and ReadMirrored settings.
     TSelf& ReadMirrored(TString database);
 
+    bool IsReadMirroredEnabled() {
+        return ReadMirroredEnabled;
+    }
+
+    auto GetDatabasesToReadFrom() {
+        return DatabasesToReadFrom;
+    }
+
+private:
     // Read policy settings, set via helpers above
     bool ReadMirroredEnabled = false;
-    std::unordered_set<TString> Databases;
+    std::unordered_set<TString> DatabasesToReadFrom;
 };
 
 

ydb/public/sdk/cpp/client/ydb_federated_topic/impl/federated_read_session.cpp

@@ -64,6 +64,7 @@ TFederatedReadSessionImpl::TFederatedReadSessionImpl(const TFederatedReadSession
     , Observer(std::move(observer))
     , AsyncInit(Observer->WaitForFirstState())
     , FederationState(nullptr)
+    , Log(Connections->GetLog())
     , SessionId(CreateGuidAsString())
 {
 }
@@ -101,15 +102,17 @@ void TFederatedReadSessionImpl::OnFederatedStateUpdateImpl() {
         CloseImpl();
         return;
     }
-    if (Settings.ReadMirroredEnabled) {
-        Y_ABORT_UNLESS(Settings.Databases.size() == 1);
-        // add -mirrored-from- topics to Settings
+    if (Settings.IsReadMirroredEnabled()) {
+        Y_ABORT_UNLESS(Settings.GetDatabasesToReadFrom().size() == 1);
+        auto dbToReadFrom = *Settings.GetDatabasesToReadFrom().begin();
 
-        // how to get mirrors in general case???
-        std::vector<TString> dcNames = {"sas", "vla", "klg", "vlx"};
+        std::vector<TString> dcNames = GetAllFederationLocations();
         auto topics = Settings.Topics_;
         for (const auto& topic : topics) {
             for (const auto& dc : dcNames) {
+                if (AsciiEqualsIgnoreCase(dc, dbToReadFrom)) {
+                    continue;
+                }
                 auto mirroredTopic = topic;
                 mirroredTopic.PartitionIds_.clear();
                 mirroredTopic.Path(topic.Path_ + "-mirrored-from-" + dc);
@@ -134,17 +137,25 @@ void TFederatedReadSessionImpl::OnFederatedStateUpdateImpl() {
     OpenSubSessionsImpl(databases);
 }
 
+std::vector<TString> TFederatedReadSessionImpl::GetAllFederationLocations() {
+    std::vector<TString> result;
+    for (const auto& db : FederationState->DbInfos) {
+        result.push_back(db->location());
+    }
+    return result;
+}
+
 bool TFederatedReadSessionImpl::IsDatabaseEligibleForRead(const std::shared_ptr<TDbInfo>& db) {
     if (db->status() != TDbInfo::Status::DatabaseInfo_Status_AVAILABLE &&
         db->status() != TDbInfo::Status::DatabaseInfo_Status_READ_ONLY) {
         return false;
     }
 
-    if (Settings.Databases.empty()) {
+    if (Settings.GetDatabasesToReadFrom().empty()) {
         return true;
     }
 
-    for (const auto& dbFromSettings : Settings.Databases) {
+    for (const auto& dbFromSettings : Settings.GetDatabasesToReadFrom()) {
         if (AsciiEqualsIgnoreCase(db->name(), dbFromSettings) ||
             AsciiEqualsIgnoreCase(db->id(), dbFromSettings)) {
             return true;

ydb/public/sdk/cpp/client/ydb_federated_topic/impl/federated_read_session.h

@@ -46,13 +46,14 @@ class TFederatedReadSessionImpl : public NPersQueue::TEnableSelfContext<TFederat
     }
 
 private:
-    // TODO logging
     TStringBuilder GetLogPrefix() const;
 
     void Start();
     bool ValidateSettings();
     void OpenSubSessionsImpl(const std::vector<std::shared_ptr<TDbInfo>>& dbInfos);
 
+    std::vector<TString> GetAllFederationLocations();
+
     bool IsDatabaseEligibleForRead(const std::shared_ptr<TDbInfo>& db);
 
     void OnFederatedStateUpdateImpl();
@@ -70,8 +71,7 @@ class TFederatedReadSessionImpl : public NPersQueue::TEnableSelfContext<TFederat
     NThreading::TFuture<void> AsyncInit;
     std::shared_ptr<TFederatedDbState> FederationState;
 
-    // TODO
-    // TLog Log;
+    TLog Log;
 
     const TString SessionId;
     const TInstant StartSessionTime = TInstant::Now();

ydb/public/sdk/cpp/client/ydb_federated_topic/impl/federated_topic.cpp

@@ -23,14 +23,17 @@ TReadOriginalSettings& TReadOriginalSettings::AddLocal() {
 }
 
 TFederatedReadSessionSettings& TFederatedReadSessionSettings::ReadOriginal(TReadOriginalSettings settings) {
-    std::swap(Databases, settings.Databases);
+    std::swap(DatabasesToReadFrom, settings.Databases);
     ReadMirroredEnabled = false;
     return *this;
 }
 
 TFederatedReadSessionSettings& TFederatedReadSessionSettings::ReadMirrored(TString database) {
-    Databases.clear();
-    Databases.insert(std::move(database));
+    if (database == "_local") {
+        ythrow TContractViolation("Reading from local database not supported, use specific database");
+    }
+    DatabasesToReadFrom.clear();
+    DatabasesToReadFrom.insert(std::move(database));
     ReadMirroredEnabled = true;
     return *this;
 }