[*] cache for SchemeNavigate

Author: Alek5andr-Kotov

ydb/core/kqp/session_actor/kqp_session_actor.cpp

@@ -550,7 +550,6 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
     }
 
     void AddOffsetsToTransaction() {
-        LOG_I("begin request for TopicOperations");
         YQL_ENSURE(QueryState);
         if (!PrepareQueryTransaction()) {
             return;
@@ -561,7 +560,6 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         if (!AreAllTheTopicsAndPartitionsKnown()) {
             auto navigate = QueryState->BuildSchemeCacheNavigate();
             Become(&TKqpSessionActor::ExecuteState);
-            LOG_I("begin request for SchemeNavigate");
             Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
             return;
         }
@@ -577,10 +575,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
             return;
         }
 
-        LOG_I("end request for TopicOperations");
         ReplySuccess();
-
-        LOG_I("after ReplySuccess");
     }
 
     void CompileQuery() {
@@ -2917,10 +2912,8 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
     }
 
     void ProcessTopicOps(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
-        LOG_I("end request for SchemeNavigate");
         YQL_ENSURE(ev->Get()->Request);
         if (ev->Get()->Request->Cookie < QueryId) {
-            LOG_I("unexpected return #2");
             return;
         }
 
@@ -2945,21 +2938,18 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
             ythrow TRequestFail(Ydb::StatusIds::BAD_REQUEST) << message;
         }
 
+        QueryState->TxCtx->TopicOperations.CacheSchemeCacheNavigate(response->ResultSet);
+
         if (HasTopicWriteOperations() && !HasTopicWriteId()) {
-            LOG_I("begin request for WriteId");
             Send(MakeTxProxyID(), new TEvTxUserProxy::TEvAllocateTxId, 0, QueryState->QueryId);
-        } else {
-            LOG_I("end request for TopicOperations");
-            ReplySuccess();
-
-            LOG_I("after ReplySuccess");
+            return;
         }
+
+        ReplySuccess();
     }
 
     void Handle(TEvTxUserProxy::TEvAllocateTxIdResult::TPtr& ev) {
-        LOG_I("end request for WriteId");
         if (CurrentStateFunc() != &TThis::ExecuteState || ev->Cookie < QueryId) {
-            LOG_I("unexpected return #1");
             return;
         }
 
@@ -2968,11 +2958,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
 
         SetTopicWriteId(NLongTxService::TLockHandle(ev->Get()->TxId, TActivationContext::ActorSystem()));
 
-        LOG_I("end request for TopicOperations");
         ReplySuccess();
-
-        LOG_I("current state: " << CurrentStateFuncName());
-        LOG_I("after ReplySuccess");
     }
 
     bool HasTopicWriteOperations() const {

ydb/core/kqp/topics/kqp_topics.cpp

@@ -198,6 +198,8 @@ void TTopicPartitionOperations::Merge(const TTopicPartitionOperations& rhs)
     if (Topic_.Empty()) {
         Topic_ = rhs.Topic_;
         Partition_ = rhs.Partition_;
+    }
+    if (TabletId_.Empty()) {
         TabletId_ = rhs.TabletId_;
     }
 
@@ -412,8 +414,6 @@ bool TTopicOperations::ProcessSchemeCacheNavigate(const NSchemeCache::TSchemeCac
                     p->second.SetTabletId(partition.GetTabletId());
                 }
             }
-
-            CachedNavigateResult_[path] = result;
         } else {
             builder << "Topic '" << JoinPath(result.Path) << "' is missing";
 
@@ -460,6 +460,20 @@ bool TTopicOperations::HasThisPartitionAlreadyBeenAdded(const TString& topicPath
     return false;
 }
 
+void TTopicOperations::CacheSchemeCacheNavigate(const NSchemeCache::TSchemeCacheNavigate::TResultSet& results)
+{
+    for (const auto& result : results) {
+        if (result.Kind != NSchemeCache::TSchemeCacheNavigate::KindTopic) {
+            continue;
+        }
+        if (!result.PQGroupInfo) {
+            continue;
+        }
+        TString path = CanonizePath(result.Path);
+        CachedNavigateResult_[path] = result;
+    }
+}
+
 void TTopicOperations::BuildTopicTxs(TTopicOperationTransactions& txs)
 {
     for (auto& [_, operations] : Operations_) {

ydb/core/kqp/topics/kqp_topics.h

@@ -147,6 +147,7 @@ class TTopicOperations {
     bool ProcessSchemeCacheNavigate(const NSchemeCache::TSchemeCacheNavigate::TResultSet& results,
                                     Ydb::StatusIds_StatusCode& status,
                                     TString& message);
+    void CacheSchemeCacheNavigate(const NSchemeCache::TSchemeCacheNavigate::TResultSet& results);
 
     void BuildTopicTxs(TTopicOperationTransactions &txs);