Merge 41c9e98 into bd147a9

Author: vladl2802

ydb/core/kqp/proxy_service/kqp_proxy_service.cpp

@@ -44,7 +44,6 @@
 #include <library/cpp/lwtrace/mon/mon_lwtrace.h>
 #include <library/cpp/monlib/service/pages/templates.h>
 #include <library/cpp/resource/resource.h>
-#include <util/generic/guid.h>
 
 namespace NKikimr::NKqp {
 
@@ -236,7 +235,7 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
         if (auto& cfg = TableServiceConfig.GetSpillingServiceConfig().GetLocalFileConfig(); cfg.GetEnable()) {
             TString spillingRoot = cfg.GetRoot();
             if (spillingRoot.empty()) {
-                spillingRoot = TStringBuilder() << "/tmp/ydb_spilling_" << CreateGuidAsString() << "/";
+                spillingRoot = NYql::NDq::GetTmpSpillingRootForCurrentUser();
             }
 
             SpillingService = TlsActivationContext->ExecutorThread.RegisterActor(NYql::NDq::CreateDqLocalFileSpillingService(

ydb/library/yql/dq/actors/spilling/spilling_file.cpp

@@ -13,6 +13,11 @@
 #include <util/folder/path.h>
 #include <util/stream/file.h>
 #include <util/thread/pool.h>
+#include <util/generic/guid.h>
+#include <util/folder/iterator.h>
+#include <util/generic/vector.h>
+#include <util/folder/dirut.h>
+#include <util/system/user.h>
 
 namespace NYql::NDq {
 
@@ -159,6 +164,7 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             EvCloseFileResponse = TEvDqSpillingLocalFile::EEv::LastEvent + 1,
             EvWriteFileResponse,
             EvReadFileResponse,
+            EvRemoveOldTmp,
 
             LastEvent
         };
@@ -189,6 +195,15 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             bool Removed = false;
             TMaybe<TString> Error;
         };
+
+        struct TEvRemoveOldTmp : public TEventLocal<TEvRemoveOldTmp, EvRemoveOldTmp> {
+            TFsPath TmpRoot;
+            ui32 NodeId;
+            TString SpillingSessionId;
+
+            TEvRemoveOldTmp(TFsPath tmpRoot, ui32 nodeId, TString spillingSessionId) 
+                : TmpRoot(std::move(tmpRoot)), NodeId(nodeId), SpillingSessionId(std::move(spillingSessionId)) {}
+        };
     };
 
     struct TFileDesc;
@@ -206,8 +221,12 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
 
     void Bootstrap() {
         Root_ = Config_.Root;
-        Root_ /= (TStringBuilder() << "node_" << SelfId().NodeId());
+        const auto rootToRemoveOldTmp = Root_;
+        const auto sessionId = Config_.SpillingSessionId;
+        const auto nodeId = SelfId().NodeId();
 
+        Root_ /= (TStringBuilder() << "node_" << nodeId << "_" << sessionId);
+        // Cerr << "Root from config: " << Config_.Root << ", actual root: " << Root_ << "\n";
         LOG_I("Init DQ local file spilling service at " << Root_ << ", actor: " << SelfId());
 
         try {
@@ -221,6 +240,8 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             Become(&TDqLocalFileSpillingService::BrokenState);
             return;
         }
+        
+        Send(SelfId(), MakeHolder<TEvPrivate::TEvRemoveOldTmp>(Root_, nodeId, sessionId));
 
         Become(&TDqLocalFileSpillingService::WorkState);
     }
@@ -271,6 +292,7 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
         hFunc(TEvPrivate::TEvWriteFileResponse, HandleWork)
         hFunc(TEvDqSpilling::TEvRead, HandleWork)
         hFunc(TEvPrivate::TEvReadFileResponse, HandleWork)
+        hFunc(TEvPrivate::TEvRemoveOldTmp, HandleWork)
         hFunc(NMon::TEvHttpInfo, HandleWork)
         cFunc(TEvents::TEvPoison::EventType, PassAway)
     );
@@ -712,6 +734,55 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
         Send(ev->Sender, new NMon::TEvHttpInfoRes(s.Str()));
     }
 
+    void HandleWork(TEvPrivate::TEvRemoveOldTmp::TPtr& ev) {
+        const auto& msg = *ev->Get();
+        const auto& root = msg.TmpRoot;
+        const auto nodeIdString = ToString(msg.NodeId);
+        const auto& sessionId = msg.SpillingSessionId;
+
+        LOG_I("[RemoveOldTmp] removing at root: " << root);
+
+        static const auto isDirOldTmp = [&nodeIdString, &sessionId](TString dirName) -> bool {
+            // dirName: node_<nodeId>_<sessionId>
+            static constexpr size_t NodeIdBegin = 5;
+            if (dirName.Size() < NodeIdBegin || dirName.substr(0, NodeIdBegin) != "node_") {
+                return false;
+            }  
+            const auto nodeIdEnd = dirName.find('_', NodeIdBegin);
+            if (nodeIdEnd == TString::npos || dirName.substr(NodeIdBegin, nodeIdEnd - NodeIdBegin) != nodeIdString) {
+                return false;
+            }
+            if (dirName.substr(nodeIdEnd + 1) == sessionId) {
+                return false;
+            }
+            return true;
+        };
+
+        try {
+            TDirIterator iter(root, TDirIterator::TOptions().SetMaxLevel(1));
+            
+            TVector<TString> oldTmps;
+            for (const auto &dirEntry : iter) {
+                if (dirEntry.fts_info == FTS_DP) {
+                    // skip postorder visit
+                    continue;
+                }
+                
+                const auto dirName = dirEntry.fts_name;
+                if (isDirOldTmp(dirName)) {
+                    LOG_D("[RemoveOldTmp] found old temporary at " << (root / dirName));
+                    oldTmps.emplace_back(std::move(dirName));
+                }
+            }
+
+            ForEach(oldTmps.begin(), oldTmps.end(), [&root](const auto& dirName) {
+                (root / dirName).ForceDelete();
+            });
+        } catch (const yexception& e) {
+            LOG_E("[RemoveOldTmp] removing failed due to: " << e.what());
+        }
+    }
+
 private:
     void RunOp(TStringBuf opName, THolder<IObjectInQueue> op, TFileDesc& fd) {
         if (fd.HasActiveOp) {
@@ -952,6 +1023,13 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
 
 } // anonymous namespace
 
+TFsPath GetTmpSpillingRootForCurrentUser() {
+    auto root = TFsPath{GetSystemTempDir()};
+    root /= "spilling-tmp-" + GetUsername();
+    MakeDirIfNotExist(root);
+    return root;
+}
+
 IActor* CreateDqLocalFileSpillingActor(TTxId txId, const TString& details, const TActorId& client,
     bool removeBlobsAfterRead)
 {

ydb/library/yql/dq/actors/spilling/spilling_file.h

@@ -7,11 +7,14 @@
 
 #include <util/system/types.h>
 #include <util/generic/strbuf.h>
+#include <util/folder/path.h>
+#include <util/generic/guid.h>
 
 namespace NYql::NDq {
 
 struct TFileSpillingServiceConfig {
     TString Root;
+    TString SpillingSessionId = CreateGuidAsString();
     ui64 MaxTotalSize = 0;
     ui64 MaxFileSize = 0;
     ui64 MaxFilePartSize = 0;
@@ -26,6 +29,8 @@ inline NActors::TActorId MakeDqLocalFileSpillingServiceID(ui32 nodeId) {
     return NActors::TActorId(nodeId, TStringBuf(name, 12));
 }
 
+TFsPath GetTmpSpillingRootForCurrentUser();
+
 NActors::IActor* CreateDqLocalFileSpillingActor(TTxId txId, const TString& details, const NActors::TActorId& client, bool removeBlobsAfterRead);
 
 NActors::IActor* CreateDqLocalFileSpillingService(const TFileSpillingServiceConfig& config, TIntrusivePtr<TSpillingCounters> counters);

ydb/library/yql/dq/actors/spilling/spilling_file_ut.cpp

@@ -48,13 +48,19 @@ class TTestActorRuntime: public TTestActorRuntimeBase {
         return str;
     }
 
+    const TString& GetSpillingSessionId() const {
+        return SpillingSessionId_;
+    }
+
     TActorId StartSpillingService(ui64 maxTotalSize = 1000, ui64 maxFileSize = 500,
         ui64 maxFilePartSize = 100, const TFsPath& root = TFsPath::Cwd() / GetSpillingPrefix())
     {
         SpillingRoot_ = root;
+        SpillingSessionId_ = CreateGuidAsString();
 
         auto config = TFileSpillingServiceConfig{
             .Root = root.GetPath(),
+            .SpillingSessionId = SpillingSessionId_,
             .MaxTotalSize = maxTotalSize,
             .MaxFileSize = maxFileSize,
             .MaxFilePartSize = maxFilePartSize
@@ -91,6 +97,7 @@ class TTestActorRuntime: public TTestActorRuntimeBase {
 
 private:
     TFsPath SpillingRoot_;
+    TString SpillingSessionId_;
 };
 
 TBuffer CreateBlob(ui32 size, char symbol) {
@@ -303,8 +310,7 @@ Y_UNIT_TEST_SUITE(DqSpillingFileTests) {
         auto spillingActor = runtime.StartSpillingActor(tester);
 
         runtime.WaitBootstrap();
-
-        const TString filePrefix = TStringBuilder() << runtime.GetSpillingRoot().GetPath() << "/node_" << runtime.GetNodeId() << "/1_test_";
+        const TString filePrefix = TStringBuilder() << runtime.GetSpillingRoot().GetPath() << "/node_" << runtime.GetNodeId() << "_" << runtime.GetSpillingSessionId() << "/1_test_";
 
         for (ui32 i = 0; i < 5; ++i) {
             // Cerr << "---- store blob #" << i << Endl;
@@ -346,7 +352,7 @@ Y_UNIT_TEST_SUITE(DqSpillingFileTests) {
 
         runtime.WaitBootstrap();
 
-        const TString filePrefix = TStringBuilder() << runtime.GetSpillingRoot().GetPath() << "/node_" << runtime.GetNodeId() << "/1_test_";
+        const TString filePrefix = TStringBuilder() << runtime.GetSpillingRoot().GetPath() << "/node_" << runtime.GetNodeId() << "_" << runtime.GetSpillingSessionId() << "/1_test_";
 
         for (ui32 i = 0; i < 5; ++i) {
             // Cerr << "---- store blob #" << i << Endl;
@@ -393,8 +399,7 @@ Y_UNIT_TEST_SUITE(DqSpillingFileTests) {
             auto resp = runtime.GrabEdgeEvent<TEvDqSpilling::TEvWriteResult>(tester);
             UNIT_ASSERT_VALUES_EQUAL(0, resp->Get()->BlobId);
         }
-
-        auto nodePath = TFsPath("node_" + std::to_string(spillingSvc.NodeId()));
+        auto nodePath = TFsPath("node_" + std::to_string(spillingSvc.NodeId()) + "_" + runtime.GetSpillingSessionId());
         (runtime.GetSpillingRoot() / nodePath / "1_test_0").ForceDelete();
 
         {

ydb/library/yql/providers/dq/local_gateway/yql_dq_gateway_local.cpp

@@ -80,11 +80,9 @@ class TLocalServiceHolder {
             TActorSetupCmd(resman, TMailboxType::Simple, 0));
 
         if (withSpilling) {
-            char tempDir[MAX_PATH];
-            if (MakeTempDir(tempDir, nullptr) != 0)
-                ythrow yexception() << "LocalServiceHolder: Can't create temporary directory " << tempDir;
+            auto tempDir = NDq::GetTmpSpillingRootForCurrentUser();
 
-            auto spillingActor = NDq::CreateDqLocalFileSpillingService(NDq::TFileSpillingServiceConfig{.Root = tempDir, .CleanupOnShutdown = true}, MakeIntrusive<NDq::TSpillingCounters>(lwmGroup));
+            auto spillingActor = NDq::CreateDqLocalFileSpillingService(NDq::TFileSpillingServiceConfig{.Root = tempDir, .CleanupOnShutdown = false}, MakeIntrusive<NDq::TSpillingCounters>(lwmGroup));
 
             ServiceNode->AddLocalService(
                 NDq::MakeDqLocalFileSpillingServiceID(nodeId),