[*] unnecessary SchemeCache calls

Author: Alek5andr-Kotov

ydb/core/kqp/session_actor/kqp_session_actor.cpp

@@ -531,6 +531,21 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         CompileQuery();
     }
 
+    bool AreAllTheTopicsAndPartitionsKnown() const {
+        const NKikimrKqp::TTopicOperationsRequest& operations = QueryState->GetTopicOperations();
+        for (const auto& topic : operations.GetTopics()) {
+            auto path = CanonizePath(NPersQueue::GetFullTopicPath(TlsActivationContext->AsActorContext(),
+                                                                  QueryState->GetDatabase(), topic.path()));
+
+            for (const auto& partition : topic.partitions()) {
+                if (!QueryState->TxCtx->TopicOperations.HasThisPartitionAlreadyBeenAdded(path, partition.partition_id())) {
+                    return false;
+                }
+            }
+        }
+        return true;
+    }
+
     void AddOffsetsToTransaction() {
         YQL_ENSURE(QueryState);
         if (!PrepareQueryTransaction()) {
@@ -539,10 +554,25 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
 
         QueryState->AddOffsetsToTransaction();
 
-        auto navigate = QueryState->BuildSchemeCacheNavigate();
+        if (!AreAllTheTopicsAndPartitionsKnown()) {
+            auto navigate = QueryState->BuildSchemeCacheNavigate();
+            Become(&TKqpSessionActor::ExecuteState);
+            Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
+            return;
+        }
 
-        Become(&TKqpSessionActor::ExecuteState);
-        Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
+        TString message;
+        if (!QueryState->TryMergeTopicOffsets(QueryState->TopicOperations, message)) {
+            ythrow TRequestFail(Ydb::StatusIds::BAD_REQUEST) << message;
+        }
+
+        if (HasTopicWriteOperations() && !HasTopicWriteId()) {
+            Become(&TKqpSessionActor::ExecuteState);
+            Send(MakeTxProxyID(), new TEvTxUserProxy::TEvAllocateTxId, 0, QueryState->QueryId);
+            return;
+        }
+
+        ReplySuccess();
     }
 
     void CompileQuery() {

ydb/core/kqp/topics/kqp_topics.cpp

@@ -194,7 +194,6 @@ void TTopicPartitionOperations::Merge(const TTopicPartitionOperations& rhs)
 {
     Y_ABORT_UNLESS(Topic_.Empty() || Topic_ == rhs.Topic_);
     Y_ABORT_UNLESS(Partition_.Empty() || Partition_ == rhs.Partition_);
-    Y_ABORT_UNLESS(TabletId_.Empty() || TabletId_ == rhs.TabletId_);
 
     if (Topic_.Empty()) {
         Topic_ = rhs.Topic_;
@@ -429,6 +428,11 @@ bool TTopicOperations::ProcessSchemeCacheNavigate(const NSchemeCache::TSchemeCac
     return true;
 }
 
+bool TTopicOperations::HasThisPartitionAlreadyBeenAdded(const TString& topic, ui32 partitionId) const
+{
+    return Operations_.contains({topic, partitionId});
+}
+
 void TTopicOperations::BuildTopicTxs(TTopicOperationTransactions& txs)
 {
     for (auto& [_, operations] : Operations_) {

ydb/core/kqp/topics/kqp_topics.h

@@ -159,6 +159,8 @@ class TTopicOperations {
 
     size_t GetSize() const;
 
+    bool HasThisPartitionAlreadyBeenAdded(const TString& topic, ui32 partitionId) const;
+
 private:
     THashMap<TTopicPartition, TTopicPartitionOperations, TTopicPartition::THash> Operations_;
     bool HasReadOperations_ = false;