Early cleanup of uncommited qstorage files (#4294)

4 files changed, 28 insertions, 9 deletions

diff --git a/ydb/library/yql/core/qplayer/storage/file/yql_qstorage_file.cpp b/ydb/library/yql/core/qplayer/storage/file/yql_qstorage_file.cpp
index ddd7fa2c5c7..b5ec89152bd 100644
--- a/ydb/library/yql/core/qplayer/storage/file/yql_qstorage_file.cpp
+++ b/ydb/library/yql/core/qplayer/storage/file/yql_qstorage_file.cpp
@@ -19,7 +19,11 @@ protected:
     TWriterBase(TFsPath& path, TInstant writtenAt)
         : Path_(path)
         , WrittenAt_(writtenAt)
-    {}
+    {
+        NFs::Remove(Path_.GetPath() + ".dat");
+        NFs::Remove(Path_.GetPath() + ".idx");
+        NFs::Remove(Path_.GetPath() + ".idx.tmp");
+    }
 
 protected:
     void WriteIndex(ui64 totalItems, ui64 totalBytes, ui64 checksum) const {
@@ -100,9 +104,16 @@ public:
     TUnbufferedWriter(TFsPath& path, TInstant writtenAt, const TQWriterSettings& settings)
         : TWriterBase(path, writtenAt)
         , Settings_(settings)
-        , DataFile_(Path_.GetPath() + ".dat")
     {
-        DataFile_.Write(&WrittenAt_, sizeof(WrittenAt_));
+        DataFile_.ConstructInPlace(Path_.GetPath() + ".dat");
+        DataFile_->Write(&WrittenAt_, sizeof(WrittenAt_));
+    }
+
+    ~TUnbufferedWriter() {
+        if (!Committed_) {
+            DataFile_.Clear();
+            NFs::Remove(Path_.GetPath() + ".dat");
+        }
     }
 
     NThreading::TFuture<void> Put(const TQItemKey& key, const TString& value) final {
@@ -110,9 +121,9 @@ public:
             Y_ENSURE(!Committed_);
             if (!Overflow_) {
                 if (Keys_.emplace(key).second) {
-                    SaveString(DataFile_, key.Component, TotalBytes_, Checksum_);
-                    SaveString(DataFile_, key.Label, TotalBytes_, Checksum_);
-                    SaveString(DataFile_, value, TotalBytes_, Checksum_);
+                    SaveString(*DataFile_, key.Component, TotalBytes_, Checksum_);
+                    SaveString(*DataFile_, key.Label, TotalBytes_, Checksum_);
+                    SaveString(*DataFile_, value, TotalBytes_, Checksum_);
                     ++TotalItems_;
                 }
 
@@ -137,7 +148,8 @@ public:
 
             Y_ENSURE(!Committed_);
             Committed_ = true;
-            DataFile_.Finish();
+            DataFile_->Finish();
+            DataFile_.Clear();
             WriteIndex(TotalItems_, TotalBytes_, Checksum_);
             return NThreading::MakeFuture();
         }
@@ -146,7 +158,7 @@ public:
 private:
     const TQWriterSettings Settings_;
     TMutex Mutex_;
-    TFileOutput DataFile_;
+    TMaybe<TFileOutput> DataFile_;
     ui64 TotalItems_ = 0;
     ui64 TotalBytes_ = 0;
     ui64 Checksum_ = 0;
diff --git a/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.cpp b/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.cpp
index d62efc7ebfb..27ad7978898 100644
--- a/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.cpp
+++ b/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.cpp
@@ -23,6 +23,11 @@ void QStorageTestEmpty_Impl(const NYql::IQStoragePtr& storage) {
     UNIT_ASSERT(!iterator->Next().GetValueSync().Defined());
 }
 
+void QStorageTestNoCommit_Impl(const NYql::IQStoragePtr& storage) {
+    auto writer = storage->MakeWriter("foo", {});
+    writer->Put({"comp", "label"}, "value").GetValueSync();
+}
+
 void QStorageTestOne_Impl(const NYql::IQStoragePtr& storage) {
     auto writer = storage->MakeWriter("foo", {});
     writer->Put({"comp", "label"}, "value").GetValueSync();
diff --git a/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.h b/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.h
index 7c6108b3926..cb787b36baa 100644
--- a/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.h
+++ b/ydb/library/yql/core/qplayer/storage/ut_common/yql_qstorage_ut_common.h
@@ -4,6 +4,7 @@
 #include <library/cpp/testing/unittest/registar.h>
 
 void QStorageTestEmpty_Impl(const NYql::IQStoragePtr& storage);
+void QStorageTestNoCommit_Impl(const NYql::IQStoragePtr& storage);
 void QStorageTestOne_Impl(const NYql::IQStoragePtr& storage);
 void QStorageTestManyKeys_Impl(const NYql::IQStoragePtr& storage);
 void QStorageTestInterleaveReadWrite_Impl(const NYql::IQStoragePtr& storage);
@@ -20,6 +21,7 @@ void QStorageTestLimitWriterBytes_Impl(const NYql::IQStoragePtr& storage);
 
 #define GENERATE_TESTS(FACTORY)\
     GENERATE_ONE_TEST(Empty, FACTORY) \
+    GENERATE_ONE_TEST(NoCommit, FACTORY) \
     GENERATE_ONE_TEST(One, FACTORY) \
     GENERATE_ONE_TEST(ManyKeys, FACTORY) \
     GENERATE_ONE_TEST(InterleaveReadWrite, FACTORY) \
diff --git a/ydb/library/yql/tools/dqrun/dqrun.cpp b/ydb/library/yql/tools/dqrun/dqrun.cpp
index 528d21efeef..d2ac2c0d469 100644
--- a/ydb/library/yql/tools/dqrun/dqrun.cpp
+++ b/ydb/library/yql/tools/dqrun/dqrun.cpp
@@ -1065,7 +1065,7 @@ int RunMain(int argc, const char* argv[])
         SerializeToTextFormat(snapshot, *output.Get());
     }
 
-    if (res.Has("capture")) {
+    if (result == 0 && res.Has("capture")) {
         qContext.GetWriter()->Commit().GetValueSync();
     }