Merge 8b1dd39 into ae7146c

Author: vladl2802

ydb/core/kqp/proxy_service/kqp_proxy_service.cpp

@@ -44,8 +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 {
 
 namespace {
@@ -236,7 +234,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 GuidString;
+
+            TEvRemoveOldTmp(TFsPath tmpRoot, ui32 nodeId, TString guidString) 
+                : TmpRoot(std::move(tmpRoot)), NodeId(nodeId), GuidString(std::move(guidString)) {}
+        };
     };
 
     struct TFileDesc;
@@ -206,8 +221,13 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
 
     void Bootstrap() {
         Root_ = Config_.Root;
-        Root_ /= (TStringBuilder() << "node_" << SelfId().NodeId());
+        const auto nodeId = SelfId().NodeId();
+        const auto guidString = TGUID::Create().AsGuidString();
 
+        Send(SelfId(), MakeHolder<TEvPrivate::TEvRemoveOldTmp>(Root_, nodeId, guidString));
+
+        Root_ /= (TStringBuilder() << "node_" << nodeId << "_" << guidString);
+        Cerr << "Root from config: " << Config_.Root << ", actual root: " << Root_ << "\n";
         LOG_I("Init DQ local file spilling service at " << Root_ << ", actor: " << SelfId());
 
         try {
@@ -271,6 +291,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 +733,41 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
         Send(ev->Sender, new NMon::TEvHttpInfoRes(s.Str()));
     }
 
+    void HandleWork(TEvPrivate::TEvRemoveOldTmp::TPtr& ev) {
+        auto& msg = *ev->Get();
+        auto& root = msg.TmpRoot;
+        auto nodeIdString = ToString(msg.NodeId);
+        auto& guidString = msg.GuidString;
+
+        LOG_I("[RemoveOldTmp] removing at root: " << root);
+
+        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;
+                }
+                TString dirName = dirEntry.fts_name;
+                TVector<TString> parts;
+                StringSplitter(dirName).Split('_').Collect(&parts);
+                
+                if (parts.size() == 3 && parts[0] == "node" && parts[1] == nodeIdString && parts[2] != guidString) {
+                    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 +1008,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,6 +7,7 @@
 
 #include <util/system/types.h>
 #include <util/generic/strbuf.h>
+#include <util/folder/path.h>
 
 namespace NYql::NDq {
 
@@ -26,6 +27,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/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),