fixes applied to sessions sysview (#8667)

Author: gridnevvvit

ydb/core/kqp/proxy_service/kqp_proxy_service.cpp

@@ -1732,7 +1732,11 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
         KQP_PROXY_LOG_D("incoming list sessions request " << ev->Get()->Record.ShortUtf8DebugString());
 
         auto result = std::make_unique<TEvKqp::TEvListSessionsResponse>();
-        auto startIt = LocalSessions->GetOrderedLowerBound(ev->Get()->Record.GetSessionIdStart());
+
+        const auto& tenant = ev->Get()->Record.GetTenantName();
+        bool checkTenant = (AppData()->TenantName != tenant);
+
+        auto startIt = LocalSessions->GetOrderedLowerBound(tenant, ev->Get()->Record.GetSessionIdStart());
         auto endIt = LocalSessions->GetOrderedEnd();
         i32 freeSpace = ev->Get()->Record.GetFreeSpace();
 
@@ -1743,6 +1747,10 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
 
         while(startIt != endIt && freeSpace > 0) {
             auto* sessionInfo = startIt->second;
+            if (checkTenant && sessionInfo->Database != ev->Get()->Record.GetTenantName()) {
+                finished = true;
+                break;
+            }
 
             if (!until.empty()) {
                 if (sessionInfo->SessionId > until) {
@@ -1770,7 +1778,8 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
         if (finished) {
             result->Record.SetFinished(true);
         } else {
-            result->Record.SetContinuationToken(startIt->first);
+            Y_ABORT_UNLESS(startIt != endIt);
+            result->Record.SetContinuationToken(startIt->first.second);
             result->Record.SetFinished(false);
         }
 

ydb/core/kqp/proxy_service/kqp_proxy_service_impl.h

@@ -148,7 +148,7 @@ struct TKqpSessionInfo {
 
 class TLocalSessionsRegistry {
     THashMap<TString, TKqpSessionInfo> LocalSessions;
-    std::map<TString, TKqpSessionInfo*> OrderedSessions;
+    std::map<std::pair<TString, TString>, TKqpSessionInfo*> OrderedSessions;
     THashMap<TActorId, TString> TargetIdIndex;
     THashSet<TString> ShutdownInFlightSessions;
     THashMap<TString, ui32> SessionsCountPerDatabase;
@@ -206,7 +206,7 @@ class TLocalSessionsRegistry {
         auto result = LocalSessions.emplace(sessionId,
             TKqpSessionInfo(sessionId, workerId, database, dbCounters, std::move(pos),
                 sessionStartedAt + idleDuration, IdleSessions.end(), pgWire, startedAt));
-        OrderedSessions.emplace(sessionId, &result.first->second);
+        OrderedSessions.emplace(std::make_pair(database, sessionId), &result.first->second);
         SessionsCountPerDatabase[database]++;
         Y_ABORT_UNLESS(result.second, "Duplicate session id!");
         TargetIdIndex.emplace(workerId, sessionId);
@@ -300,11 +300,11 @@ class TLocalSessionsRegistry {
         return ShutdownInFlightSessions.size();
     }
 
-    std::map<TString, TKqpSessionInfo*>::const_iterator GetOrderedLowerBound(const TString& continuation) const {
-        return OrderedSessions.lower_bound(continuation);
+    std::map<std::pair<TString, TString>, TKqpSessionInfo*>::const_iterator GetOrderedLowerBound(const TString& tenant, const TString& continuation) const {
+        return OrderedSessions.lower_bound(std::make_pair(tenant, continuation));
     }
 
-    std::map<TString, TKqpSessionInfo*>::const_iterator GetOrderedEnd() const {
+    std::map<std::pair<TString, TString>, TKqpSessionInfo*>::const_iterator GetOrderedEnd() const {
         return OrderedSessions.end();
     }
 
@@ -337,7 +337,7 @@ class TLocalSessionsRegistry {
                 }
             }
 
-            OrderedSessions.erase(sessionId);
+            OrderedSessions.erase(std::make_pair(it->second.Database, sessionId));
             LocalSessions.erase(it);
         }
 

ydb/core/kqp/proxy_service/kqp_session_info.cpp

@@ -6,6 +6,8 @@ namespace NKikimr::NKqp {
 
 using VSessions = NKikimr::NSysView::Schema::QuerySessions;
 
+constexpr size_t QUERY_TEXT_LIMIT = 10_KB;
+
 void TKqpSessionInfo::SerializeTo(::NKikimrKqp::TSessionInfo* proto, const TFieldsMap& fieldsMap) const {
     if (fieldsMap.NeedField(VSessions::SessionId::ColumnId)) {  // 1
         proto->SetSessionId(SessionId);
@@ -26,7 +28,12 @@ void TKqpSessionInfo::SerializeTo(::NKikimrKqp::TSessionInfo* proto, const TFiel
 
     // last executed query or currently running query.
     if (fieldsMap.NeedField(VSessions::Query::ColumnId)) {  // 4
-        proto->SetQuery(QueryText);
+        if (QueryText.size() > QUERY_TEXT_LIMIT) {
+            TString truncatedText = QueryText.substr(0, QUERY_TEXT_LIMIT);
+            proto->SetQuery(QueryText);
+        } else {
+            proto->SetQuery(QueryText);
+        }
     }
 
     if (fieldsMap.NeedField(VSessions::QueryCount::ColumnId)) {  // 5

ydb/core/protos/kqp.proto

@@ -359,6 +359,7 @@ message TEvListSessionsRequest {
     repeated uint32 Columns = 5;
     optional int64 FreeSpace = 6;
     optional int64 Limit = 7;
+    optional string TenantName = 8;
 }
 
 message TEvListSessionsResponse {

ydb/core/sys_view/sessions/sessions.cpp

@@ -178,6 +178,7 @@ class TSessionsScan : public NKikimr::NSysView::TScanActorBase<TSessionsScan> {
             const auto& nodeId = PendingNodes.front();
             auto kqpProxyId = NKqp::MakeKqpProxyID(nodeId);
             auto req = std::make_unique<NKikimr::NKqp::TEvKqp::TEvListSessionsRequest>();
+            req->Record.SetTenantName(TenantName);
             if (!ContinuationToken.empty()) {
                 req->Record.SetSessionIdStart(ContinuationToken);
                 req->Record.SetSessionIdStartInclusive(true);