Merge f5c9f17 into b1a2b73

Author: vladl2802

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

@@ -13,6 +13,9 @@
 #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>
 
 namespace NYql::NDq {
 
@@ -159,6 +162,7 @@ class TDqLocalFileSpillingService : public TActorBootstrapped<TDqLocalFileSpilli
             EvCloseFileResponse = TEvDqSpillingLocalFile::EEv::LastEvent + 1,
             EvWriteFileResponse,
             EvReadFileResponse,
+            EvRemoveOldTmp,
 
             LastEvent
         };
@@ -189,6 +193,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,7 +219,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 << ", Root_: " << Root_ << "\n";
 
         LOG_I("Init DQ local file spilling service at " << Root_ << ", actor: " << SelfId());
 
@@ -271,6 +290,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 +732,37 @@ 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);
+
+        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();
+        });
+    }
+
 private:
     void RunOp(TStringBuf opName, THolder<IObjectInQueue> op, TFileDesc& fd) {
         if (fd.HasActiveOp) {

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

@@ -80,11 +80,12 @@ 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 = TFsPath{GetSystemTempDir()};
+            tempDir /= "spilling-service-tmp";
 
-            auto spillingActor = NDq::CreateDqLocalFileSpillingService(NDq::TFileSpillingServiceConfig{.Root = tempDir, .CleanupOnShutdown = true}, MakeIntrusive<NDq::TSpillingCounters>(lwmGroup));
+            MakeDirIfNotExist(tempDir);
+
+            auto spillingActor = NDq::CreateDqLocalFileSpillingService(NDq::TFileSpillingServiceConfig{.Root = tempDir, .CleanupOnShutdown = false}, MakeIntrusive<NDq::TSpillingCounters>(lwmGroup));
 
             ServiceNode->AddLocalService(
                 NDq::MakeDqLocalFileSpillingServiceID(nodeId),