fixed issues

evanevanevanevannnn · evanevanevanevannnn · commit 0dfe76b7c502 · 2025-03-03T21:53:27.000Z
diff --git a/ydb/core/kqp/host/kqp_host.cpp b/ydb/core/kqp/host/kqp_host.cpp
@@ -1893,6 +1893,7 @@ class TKqpHost : public IKqpHost {
         solomonState->Gateway = FederatedQuerySetup->SolomonGateway;
         solomonState->DqIntegration = NYql::CreateSolomonDqIntegration(solomonState);
         solomonState->Configuration->Init(FederatedQuerySetup->SolomonGatewayConfig, TypesCtx);
+        solomonState->ExecutorPoolId = AppData()->UserPoolId;
 
         TypesCtx->AddDataSource(NYql::SolomonProviderName, NYql::CreateSolomonDataSource(solomonState));
         TypesCtx->AddDataSink(NYql::SolomonProviderName, NYql::CreateSolomonDataSink(solomonState));
diff --git a/ydb/library/yql/providers/solomon/actors/dq_solomon_metrics_queue.cpp b/ydb/library/yql/providers/solomon/actors/dq_solomon_metrics_queue.cpp
@@ -26,6 +26,8 @@
 
 namespace NYql::NDq {
 
+namespace {
+
 class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomonMetricsQueueActor> {
 public:
     static constexpr char ActorName[] = "DQ_SOLOMON_METRICS_QUEUE_ACTOR";
@@ -75,9 +77,8 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
         , BatchCountLimit(batchCountLimit)
         , ReadParams(std::move(readParams))
         , CredentialsProvider(credentialsProvider)
-    {
-        SolomonClient = NSo::ISolomonAccessorClient::Make(ReadParams.Source, CredentialsProvider);
-    }
+        , SolomonClient(NSo::ISolomonAccessorClient::Make(readParams.Source, credentialsProvider))
+    {}
 
     void Bootstrap() {
         Schedule(PoisonTimeout, new NActors::TEvents::TEvPoison());
@@ -125,7 +126,7 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
     STATEFN(AnErrorOccurredState) {
         try {
             switch (const auto etype = ev->GetTypeRewrite()) {
-                hFunc(TEvSolomonProvider::TEvGetNextBatch, );
+                hFunc(TEvSolomonProvider::TEvGetNextBatch, HandleGetNextBatchForErrorState);
                 cFunc(TEvPrivatePrivate::EvRoundRobinStageTimeout, HandleRoundRobinStageTimeout);
                 cFunc(NActors::TEvents::TSystem::Poison, HandlePoison);
                 default:
@@ -171,7 +172,7 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
     }
 
     void HandleRoundRobinStageTimeout() {
-        LOG_T("TDqSolomonMetricsQueueActor","Handle round robin stage timeout");
+        LOG_T("TDqSolomonMetricsQueueActor", "Handle round robin stage timeout");
         if (!RoundRobinStageFinished) {
             RoundRobinStageFinished = true;
             AnswerPendingRequests();
@@ -189,12 +190,12 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
     }
 
     void HandleGetNextBatchForEmptyState(TEvSolomonProvider::TEvGetNextBatch::TPtr& ev) {
-        LOG_T("TDqSolomonMetricsQueueActor", "HandleGetNextBatchForEmptyState Giving away rest of Objects");
+        LOG_T("TDqSolomonMetricsQueueActor", "HandleGetNextBatchForEmptyState giving away rest of Objects");
         TrySendMetrics(ev->Sender, ev->Get()->Record.GetTransportMeta());
     }
 
     void HandleGetNextBatchForErrorState(TEvSolomonProvider::TEvGetNextBatch::TPtr& ev) {
-        LOG_D("TDqSolomonMetricsQueueActor", "HandleGetNextBatchForErrorState Sending issues");
+        LOG_D("TDqSolomonMetricsQueueActor", "HandleGetNextBatchForErrorState sending issues");
         Send(ev->Sender, new TEvSolomonProvider::TEvMetricsReadError(*MaybeIssues, ev->Get()->Record.GetTransportMeta()));
         TryFinish(ev->Sender, ev->Get()->Record.GetTransportMeta().GetSeqNo());
     }
@@ -206,7 +207,7 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
 
     void TransitToErrorState() {
         Y_ENSURE(MaybeIssues.Defined());
-        LOG_I("TDqSolomonMetricsQueueActor", "TransitToErrorState an error occurred sending ");
+        LOG_I("TDqSolomonMetricsQueueActor", "TransitToErrorState an error occurred, sending issues");
         AnswerPendingRequests();
         Metrics.clear();
         Become(&TDqSolomonMetricsQueueActor::AnErrorOccurredState);
@@ -306,16 +307,12 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
                 }
             }
 
-            bool isEmpty = true;
             for (const auto& [consumer, requests] : PendingRequests) {
                 if (!requests.empty()) {
-                    isEmpty = false;
+                    HasPendingRequests = true;
+                    break;
                 }
             }
-
-            if (isEmpty) {
-                HasPendingRequests = false;
-            }
         }
     }
 
@@ -372,9 +369,9 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
 
     void TryFinish(const NActors::TActorId& consumer, ui64 seqNo) {
         LOG_T("TDqSolomonMetricsQueueActor", "TryFinish from consumer " << consumer << ", " << FinishedConsumers.size() << " consumers already finished, seqNo=" << seqNo);
-        if (FinishingConsumerToLastSeqNo.contains(consumer)) {
+        if (auto it = FinishingConsumerToLastSeqNo.find(consumer); it != FinishingConsumerToLastSeqNo.end()) {
             LOG_T("TDqSolomonMetricsQueueActor", "TryFinish FinishingConsumerToLastSeqNo=" << FinishingConsumerToLastSeqNo[consumer]);
-            if (FinishingConsumerToLastSeqNo[consumer] < seqNo || SelfId().NodeId() == consumer.NodeId()) {
+            if (it->second < seqNo || SelfId().NodeId() == consumer.NodeId()) {
                 FinishedConsumers.insert(consumer);
                 if (FinishedConsumers.size() == ConsumersCount) {
                     PassAway();
@@ -395,29 +392,32 @@ class TDqSolomonMetricsQueueActor : public NActors::TActorBootstrapped<TDqSolomo
     THashSet<NActors::TActorId> FinishedConsumers;
     THashMap<NActors::TActorId, ui64> FinishingConsumerToLastSeqNo;
 
-    NSo::ISolomonAccessorClient::TPtr SolomonClient;
     TMaybe<NThreading::TFuture<NSo::ISolomonAccessorClient::TListMetricsResult>> ListingFuture;
     bool HasPendingRequests;
     THashMap<NActors::TActorId, TDeque<NDqProto::TMessageTransportMeta>> PendingRequests;
     std::vector<NSo::MetricQueue::TMetricLabels> Metrics;
     TMaybe<TString> MaybeIssues;
-
+    
     ui64 PageSize;
     ui64 PrefetchSize;
     ui64 BatchCountLimit;
     const TDqSolomonReadParams ReadParams;
     const std::shared_ptr<NYdb::ICredentialsProvider> CredentialsProvider;
+    const NSo::ISolomonAccessorClient::TPtr SolomonClient;
 
     static constexpr TDuration PoisonTimeout = TDuration::Hours(3);
     static constexpr TDuration RoundRobinStageTimeout = TDuration::Seconds(3);
 };
 
+
+} // namespace
+
 NActors::IActor* CreateSolomonMetricsQueueActor(
     ui64 consumersCount,
     TDqSolomonReadParams readParams,
     std::shared_ptr<NYdb::ICredentialsProvider> credentialsProvider)
 {
-    auto& settings = readParams.Source.settings();
+    const auto& settings = readParams.Source.settings();
 
     ui64 pageSize = 0;
     if (auto it = settings.find("metricsQueuePageSize"); it != settings.end()) {
diff --git a/ydb/library/yql/providers/solomon/actors/dq_solomon_read_actor.cpp b/ydb/library/yql/providers/solomon/actors/dq_solomon_read_actor.cpp
@@ -40,21 +40,21 @@
 #include <util/generic/hash.h>
 #include <util/system/compiler.h>
 
-#define SINK_LOG_T(s) \
+#define SOURCE_LOG_T(s) \
     LOG_TRACE_S(*NActors::TlsActivationContext, NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG_D(s) \
+#define SOURCE_LOG_D(s) \
     LOG_DEBUG_S(*NActors::TlsActivationContext, NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG_I(s) \
+#define SOURCE_LOG_I(s) \
     LOG_INFO_S(*NActors::TlsActivationContext,  NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG_W(s) \
+#define SOURCE_LOG_W(s) \
     LOG_WARN_S(*NActors::TlsActivationContext, NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG_N(s) \
+#define SOURCE_LOG_N(s) \
     LOG_NOTICE_S(*NActors::TlsActivationContext, NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG_E(s) \
+#define SOURCE_LOG_E(s) \
     LOG_ERROR_S(*NActors::TlsActivationContext, NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG_C(s) \
+#define SOURCE_LOG_C(s) \
     LOG_CRIT_S(*NActors::TlsActivationContext,  NKikimrServices::KQP_COMPUTE, LogPrefix << s)
-#define SINK_LOG(prio, s) \
+#define SOURCE_LOG(prio, s) \
     LOG_LOG_S(*NActors::TlsActivationContext, prio, NKikimrServices::KQP_COMPUTE, LogPrefix << s)
 
 namespace NYql::NDq {
@@ -87,9 +87,6 @@ auto RetryPolicy = NYql::NDq::THttpSenderRetryPolicy::GetExponentialBackoffPolic
         return ERetryErrorClass::ShortRetry;
     });
 
-} // namespace
-
-
 class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadActor>, public IDqComputeActorAsyncInput {
 public:
     static constexpr char ActorName[] = "DQ_SOLOMON_READ_ACTOR";
@@ -117,21 +114,21 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
         , MaxInflightDataRequests(maxInflightDataRequests)
         , MetricsQueueActor(metricsQueueActor)
         , CredentialsProvider(credentialsProvider)
+        , SolomonClient(NSo::ISolomonAccessorClient::Make(readParams.Source, credentialsProvider))
     {
         Y_UNUSED(counters);
-        SINK_LOG_D("Init");
+        SOURCE_LOG_D("Init");
         IngressStats.Level = statsLevel;
 
-        SolomonClient = NSo::ISolomonAccessorClient::Make(ReadParams.Source, CredentialsProvider);
         UseMetricsQueue = ReadParams.Source.HasSelectors();
 
         auto stringType = ProgramBuilder.NewDataType(NYql::NUdf::TDataType<char*>::Id);
         DictType = ProgramBuilder.NewDictType(stringType, stringType, false);
 
-        FillSystemColumnPositionindex();
+        FillSystemColumnPositionIndex();
     }
 
-    void FillSystemColumnPositionindex() {
+    void FillSystemColumnPositionIndex() {
         std::vector<TString> names(ReadParams.Source.GetSystemColumns().begin(), ReadParams.Source.GetSystemColumns().end());
         names.insert(names.end(), ReadParams.Source.GetLabelNames().begin(), ReadParams.Source.GetLabelNames().end());
         std::sort(names.begin(), names.end());
@@ -172,21 +169,21 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
         auto& batch = metricsBatch->Get()->Record;
         IsMetricsQueueEmpty = batch.GetNoMoreMetrics();
         if (IsMetricsQueueEmpty && !IsConfirmedMetricsQueueFinish) {
-            SINK_LOG_D("HandleMetricsBatch MetricsQueue empty, sending finish confirmation");
+            SOURCE_LOG_D("HandleMetricsBatch MetricsQueue empty, sending finish confirmation");
             RequestMetrics();
             IsConfirmedMetricsQueueFinish = true;
         }
 
         auto& metrics = batch.GetMetrics();
 
-        SINK_LOG_D("HandleMetricsBatch batch of size " << metrics.size());
+        SOURCE_LOG_D("HandleMetricsBatch batch of size " << metrics.size());
         Metrics.insert(Metrics.end(), metrics.begin(), metrics.end());
         ListedMetrics += metrics.size();
 
         while (TryRequestData()) {}
 
         if (LastMetricProcessed()) {
-            Send(ComputeActorId, new TEvNewAsyncInputDataArrived(InputIndex));
+            NotifyComputeActorWithData();
         }
     }
 
@@ -197,13 +194,13 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
 
         IsMetricsQueueEmpty = true;
         if (!IsConfirmedMetricsQueueFinish) {
-            SINK_LOG_D("HandleMetricsReadError sending finish confirmation to MetricsQueue");
+            SOURCE_LOG_D("HandleMetricsReadError sending finish confirmation to MetricsQueue");
             RequestMetrics();
             IsConfirmedMetricsQueueFinish = true;
         }
 
         TIssues issues { TIssue(metricsReadError->Get()->Record.GetIssues()) };
-        SINK_LOG_W("Got " << "error response[" << metricsReadError->Cookie << "] from solomon: " << issues.ToOneLineString());
+        SOURCE_LOG_W("Got " << "error response[" << metricsReadError->Cookie << "] from solomon: " << issues.ToOneLineString());
         Send(ComputeActorId, new TEvAsyncInputError(InputIndex, issues, NYql::NDqProto::StatusIds::EXTERNAL_ERROR));
         return;
     }
@@ -214,35 +211,35 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
 
         if (!batch.Success) {
             TIssues issues { TIssue(batch.ErrorMsg) };
-            SINK_LOG_W("Got " << "error response[" << newDataBatch->Cookie << "] from solomon: " << issues.ToOneLineString());
+            SOURCE_LOG_W("Got " << "error response[" << newDataBatch->Cookie << "] from solomon: " << issues.ToOneLineString());
             Send(ComputeActorId, new TEvAsyncInputError(InputIndex, issues, NYql::NDqProto::StatusIds::EXTERNAL_ERROR));
             return;
         }
 
-        SINK_LOG_D("HandleNewDataBatch new data batch");
+        SOURCE_LOG_D("HandleNewDataBatch new data batch");
         MetricsData.insert(MetricsData.end(), batch.Result.begin(), batch.Result.end());
         CompletedMetrics += batch.Result.size();
 
         NotifyComputeActorWithData();
     }
 
     void Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvRetry::TPtr&) {
-        SINK_LOG_D("Handle MetricsQueue retry");
+        SOURCE_LOG_D("Handle MetricsQueue retry");
         MetricsQueueEvents.Retry();
     }
 
     void Handle(NActors::TEvInterconnect::TEvNodeDisconnected::TPtr& ev) {
-        SINK_LOG_D("Handle MetricsQueue disconnected " << ev->Get()->NodeId);
+        SOURCE_LOG_D("Handle MetricsQueue disconnected " << ev->Get()->NodeId);
         MetricsQueueEvents.HandleNodeDisconnected(ev->Get()->NodeId);
     }
 
     void Handle(NActors::TEvInterconnect::TEvNodeConnected::TPtr& ev) {
-        SINK_LOG_D("Handle MetricsQueue connected " << ev->Get()->NodeId);
+        SOURCE_LOG_D("Handle MetricsQueue connected " << ev->Get()->NodeId);
         MetricsQueueEvents.HandleNodeConnected(ev->Get()->NodeId);
     }
 
     void Handle(NActors::TEvents::TEvUndelivered::TPtr& ev) {
-        SINK_LOG_D("Handle MetricsQueue undelivered ");
+        SOURCE_LOG_D("Handle MetricsQueue undelivered");
         if (MetricsQueueEvents.HandleUndelivered(ev) != NYql::NDq::TRetryEventsQueue::ESessionState::WrongSession) {
             TIssues issues{TIssue{TStringBuilder() << "MetricsQueue was lost"}};
             Send(ComputeActorId, new TEvAsyncInputError(InputIndex, issues, NYql::NDqProto::StatusIds::UNAVAILABLE));
@@ -252,7 +249,7 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
     i64 GetAsyncInputData(TUnboxedValueBatch& buffer, TMaybe<TInstant>&, bool& finished, i64 freeSpace) final {
         Y_UNUSED(freeSpace);
         YQL_ENSURE(!buffer.IsWide(), "Wide stream is not supported");
-        SINK_LOG_D("GetAsyncInputData sending data");
+        SOURCE_LOG_D("GetAsyncInputData sending data");
 
         for (const auto& data : MetricsData) {
             auto& labels = data.Labels;
@@ -323,7 +320,7 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
 private:
     // IActor & IDqComputeActorAsyncInput
     void PassAway() override { // Is called from Compute Actor
-        SINK_LOG_D("PassAway");
+        SOURCE_LOG_D("PassAway");
         TActor<TDqSolomonReadActor>::PassAway();
     }
 
@@ -418,7 +415,6 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
     const ui64 MaxInflightDataRequests;
 
     bool UseMetricsQueue;
-    NSo::ISolomonAccessorClient::TPtr SolomonClient;
     TRetryEventsQueue MetricsQueueEvents;
     NActors::TActorId MetricsQueueActor;
     bool IsWaitingMetricsQueueResponse = false;
@@ -433,11 +429,15 @@ class TDqSolomonReadActor : public NActors::TActorBootstrapped<TDqSolomonReadAct
 
     TString SourceId;
     std::shared_ptr<NYdb::ICredentialsProvider> CredentialsProvider;
+    NSo::ISolomonAccessorClient::TPtr SolomonClient;
     TType* DictType = nullptr;
     std::vector<size_t> SystemColumnPositionIndex;
     THashMap<TString, size_t> Index;
 };
 
+
+} // namespace
+
 std::pair<NYql::NDq::IDqComputeActorAsyncInput*, NActors::IActor*> CreateDqSolomonReadActor(
     NYql::NSo::NProto::TDqSolomonSource&& source,
     ui64 inputIndex,
diff --git a/ydb/library/yql/providers/solomon/actors/dq_solomon_write_actor.cpp b/ydb/library/yql/providers/solomon/actors/dq_solomon_write_actor.cpp
@@ -87,8 +87,6 @@ struct TMetricsInflight {
     ui64 BodySize = 0;
 };
 
-} // namespace
-
 TString GetSolomonUrl(const TString& endpoint, bool useSsl, const TString& project, const TString& cluster, const TString& service, const ::NYql::NSo::NProto::ESolomonClusterType& type) {
     TUrlBuilder builder((useSsl ? "https://" : "http://") + endpoint);
 
@@ -501,6 +499,9 @@ class TDqSolomonWriteActor : public NActors::TActor<TDqSolomonWriteActor>, publi
     ui64 Cookie = 0;
 };
 
+
+} // namespace
+
 std::pair<NYql::NDq::IDqComputeActorAsyncOutput*, NActors::IActor*> CreateDqSolomonWriteActor(
     NYql::NSo::NProto::TDqSolomonShard&& settings,
     ui64 outputIndex,
diff --git a/ydb/library/yql/providers/solomon/events/events.h b/ydb/library/yql/providers/solomon/events/events.h
@@ -36,17 +36,17 @@ struct TEvSolomonProvider {
                 metrics.begin(), 
                 metrics.end());
             Record.SetNoMoreMetrics(noMoreMetrics);
-            Record.MutableTransportMeta()->CopyFrom(transportMeta);
+            *Record.MutableTransportMeta() = transportMeta;
         }
     };
 
     struct TEvMetricsReadError:
         public NActors::TEventPB<TEvMetricsReadError, NSo::MetricQueue::TEvMetricsReadError, EvMetricsReadError> {
         
         TEvMetricsReadError() = default;
-        TEvMetricsReadError(const TString &issues, const NDqProto::TMessageTransportMeta& transportMeta) {
+        TEvMetricsReadError(const TString& issues, const NDqProto::TMessageTransportMeta& transportMeta) {
             Record.SetIssues(issues);
-            Record.MutableTransportMeta()->CopyFrom(transportMeta);
+            *Record.MutableTransportMeta() = transportMeta;
         }
     };
 
diff --git a/ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp b/ydb/library/yql/providers/solomon/provider/yql_solomon_dq_integration.cpp
@@ -329,7 +329,9 @@ class TSolomonDqIntegration: public TDqIntegrationBase {
                     maxPartitions,
                     readParams,
                     credentialsProvider
-                )
+                ),
+                NActors::TMailboxType::HTSwap,
+                State_->ExecutorPoolId
             );
 
             NActorsProto::TActorId protoId;
diff --git a/ydb/library/yql/providers/solomon/provider/yql_solomon_provider.h b/ydb/library/yql/providers/solomon/provider/yql_solomon_provider.h
@@ -28,6 +28,7 @@ struct TSolomonState : public TThrRefBase
     ISecuredServiceAccountCredentialsFactory::TPtr CredentialsFactory;
     TSolomonConfiguration::TPtr Configuration = MakeIntrusive<TSolomonConfiguration>();
     THolder<IDqIntegration> DqIntegration;
+    ui32 ExecutorPoolId = 0;
 };
 
 TDataProviderInitializer GetSolomonDataProviderInitializer(ISolomonGateway::TPtr gateway, ISecuredServiceAccountCredentialsFactory::TPtr credentialsFactory = nullptr, bool supportRtmrMode = true);
diff --git a/ydb/library/yql/providers/solomon/solomon_accessor/client/solomon_accessor_client.cpp b/ydb/library/yql/providers/solomon/solomon_accessor/client/solomon_accessor_client.cpp
@@ -365,7 +365,7 @@ ISolomonAccessorClient::Make(
     const NYql::NSo::NProto::TDqSolomonSource& source,
     std::shared_ptr<NYdb::ICredentialsProvider> credentialsProvider)
 {
-    auto& settings = source.settings();
+    const auto& settings = source.settings();
 
     TString defaultReplica = "sas";
     if (auto it = settings.find("solomonClientDefaultReplica"); it != settings.end()) {

Original file line number	Diff line number	Diff line change
`@@ -365,7 +365,7 @@ ISolomonAccessorClient::Make(`
`365`	`365`	`const NYql::NSo::NProto::TDqSolomonSource& source,`
`366`	`366`	`std::shared_ptr<NYdb::ICredentialsProvider> credentialsProvider)`
`367`	`367`	`{`
`368`		`- auto& settings = source.settings();`
	`368`	`+ const auto& settings = source.settings();`
`369`	`369`
`370`	`370`	`TString defaultReplica = "sas";`
`371`	`371`	`if (auto it = settings.find("solomonClientDefaultReplica"); it != settings.end()) {`