feat(spill): add SpillWriter, SpillReader and SpillChannelManager for spill-to-disk support

Author: dalingmeng

include/paimon/disk/file_io_channel.h

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <iomanip>
+#include <memory>
+#include <random>
+#include <sstream>
+#include <string>
+
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// A FileIOChannel represents a collection of files that belong logically to the same resource.
+/// An example is a collection of files that contain sorted runs of data from the same stream,
+/// that will later on be merged together.
+class PAIMON_EXPORT FileIOChannel {
+ private:
+    static constexpr int RANDOM_BYTES_LENGTH = 16;
+    static std::string GenerateRandomHexString(std::mt19937& random) {
+        std::uniform_int_distribution<int> dist(0, 255);
+        std::ostringstream hex_stream;
+        hex_stream << std::hex << std::setfill('0');
+        for (int i = 0; i < RANDOM_BYTES_LENGTH; ++i) {
+            hex_stream << std::setw(2) << dist(random);
+        }
+        return hex_stream.str();
+    }
+
+ public:
+    class PAIMON_EXPORT ID {
+     public:
+        explicit ID(std::string path) : path_(std::move(path)) {}
+
+        ID(const std::string& base_path, std::mt19937& random)
+            : path_(base_path + "/" + GenerateRandomHexString(random) + ".channel") {}
+
+        ID(const std::string& base_path, const std::string& prefix, std::mt19937& random)
+            : path_(base_path + "/" + prefix + "-" + GenerateRandomHexString(random) + ".channel") {
+        }
+
+        const std::string& GetPath() const {
+            return path_;
+        }
+
+        bool operator==(const ID& other) const {
+            return path_ == other.path_;
+        }
+
+        bool operator!=(const ID& other) const {
+            return !(*this == other);
+        }
+
+        struct Hash {
+            size_t operator()(const ID& id) const {
+                return std::hash<std::string>{}(id.path_);
+            }
+        };
+
+     private:
+        std::string path_;
+    };
+
+    class PAIMON_EXPORT Enumerator {
+     public:
+        Enumerator(std::string base_path, std::mt19937& random)
+            : path_(std::move(base_path)), name_prefix_(GenerateRandomHexString(random)) {}
+        std::shared_ptr<ID> Next() {
+            std::ostringstream filename;
+            filename << name_prefix_ << "." << std::setfill('0') << std::setw(6)
+                     << (local_counter_++) << ".channel";
+
+            std::string full_path = path_ + "/" + filename.str();
+            return std::make_shared<ID>(std::move(full_path));
+        }
+
+     private:
+        std::string path_;
+        std::string name_prefix_;
+        uint64_t local_counter_{0};
+    };
+};
+
+}  // namespace paimon

include/paimon/disk/io_manager.h

@@ -20,6 +20,7 @@
 #include <memory>
 #include <string>
 
+#include "paimon/disk/file_io_channel.h"
 #include "paimon/result.h"
 #include "paimon/visibility.h"
 
@@ -34,5 +35,11 @@ class PAIMON_EXPORT IOManager {
     virtual const std::string& GetTempDir() const = 0;
 
     virtual Result<std::string> GenerateTempFilePath(const std::string& prefix) const = 0;
+
+    virtual Result<std::shared_ptr<FileIOChannel::ID>> CreateChannel() = 0;
+
+    virtual Result<std::shared_ptr<FileIOChannel::ID>> CreateChannel(const std::string& prefix) = 0;
+
+    virtual Result<std::shared_ptr<FileIOChannel::Enumerator>> CreateChannelEnumerator() = 0;
 };
 }  // namespace paimon

src/paimon/CMakeLists.txt

@@ -598,6 +598,9 @@ if(PAIMON_BUILD_TESTS)
                     core/mergetree/merge_tree_writer_test.cpp
                     core/mergetree/write_buffer_test.cpp
                     core/mergetree/sorted_run_test.cpp
+                    core/mergetree/spill_channel_manager_test.cpp
+                    core/mergetree/spill_reader_test.cpp
+                    core/mergetree/spill_writer_test.cpp
                     core/migrate/file_meta_utils_test.cpp
                     core/operation/metrics/compaction_metrics_test.cpp
                     core/operation/data_evolution_file_store_scan_test.cpp

src/paimon/core/disk/io_manager.cpp

@@ -15,13 +15,17 @@
  */
 #include "paimon/disk/io_manager.h"
 
+#include <mutex>
+
 #include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/uuid.h"
 
 namespace paimon {
 class IOManagerImpl : public IOManager {
  public:
-    explicit IOManagerImpl(const std::string& tmp_dir) : tmp_dir_(tmp_dir) {}
+    explicit IOManagerImpl(const std::string& tmp_dir) : tmp_dir_(tmp_dir) {
+        random_ = std::mt19937(std::random_device{}());
+    }
 
     const std::string& GetTempDir() const override {
         return tmp_dir_;
@@ -35,9 +39,26 @@ class IOManagerImpl : public IOManager {
         return PathUtil::JoinPath(tmp_dir_, prefix + "-" + uuid + std::string(kSuffix));
     }
 
+    Result<std::shared_ptr<FileIOChannel::ID>> CreateChannel() override {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return std::make_shared<FileIOChannel::ID>(tmp_dir_, random_);
+    }
+
+    Result<std::shared_ptr<FileIOChannel::ID>> CreateChannel(const std::string& prefix) override {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return std::make_shared<FileIOChannel::ID>(tmp_dir_, prefix, random_);
+    }
+
+    Result<std::shared_ptr<FileIOChannel::Enumerator>> CreateChannelEnumerator() override {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return std::make_shared<FileIOChannel::Enumerator>(tmp_dir_, random_);
+    }
+
  private:
     static constexpr char kSuffix[] = ".channel";
     std::string tmp_dir_;
+    std::mutex mutex_;
+    std::mt19937 random_;
 };
 
 std::unique_ptr<IOManager> IOManager::Create(const std::string& tmp_dir) {

src/paimon/core/disk/io_manager_test.cpp

@@ -63,4 +63,34 @@ TEST(IOManagerTest, GenerateTempFilePathShouldBeDifferentAcrossCalls) {
     ASSERT_NE(path1, path2);
 }
 
+TEST(IOManagerTest, CreateChannelShouldReturnValidAndUniquePaths) {
+    auto tmp_dir = UniqueTestDirectory::Create();
+    std::unique_ptr<IOManager> manager = IOManager::Create(tmp_dir->Str());
+    const std::string prefix = "spill";
+
+    ASSERT_OK_AND_ASSIGN(auto channel1, manager->CreateChannel());
+    ASSERT_TRUE(StringUtils::StartsWith(channel1->GetPath(), tmp_dir->Str() + "/"));
+    ASSERT_TRUE(StringUtils::EndsWith(channel1->GetPath(), ".channel"));
+    ASSERT_EQ(PathUtil::GetName(channel1->GetPath()).size(), 32 + std::string(".channel").size());
+
+    ASSERT_OK_AND_ASSIGN(auto channel2, manager->CreateChannel(prefix));
+    ASSERT_TRUE(StringUtils::StartsWith(PathUtil::GetName(channel2->GetPath()), prefix + "-"));
+}
+
+TEST(IOManagerTest, CreateChannelEnumeratorShouldReturnSequentialAndUniquePaths) {
+    auto tmp_dir = UniqueTestDirectory::Create();
+    std::unique_ptr<IOManager> manager = IOManager::Create(tmp_dir->Str());
+
+    ASSERT_OK_AND_ASSIGN(auto enumerator, manager->CreateChannelEnumerator());
+
+    for (int i = 0; i < 10; ++i) {
+        auto channel_id = enumerator->Next();
+        ASSERT_TRUE(StringUtils::StartsWith(channel_id->GetPath(), tmp_dir->Str() + "/"));
+        std::string counter_str = std::to_string(i);
+        std::string padded_counter = std::string(6 - counter_str.size(), '0') + counter_str;
+        ASSERT_TRUE(
+            StringUtils::EndsWith(channel_id->GetPath(), "." + padded_counter + ".channel"));
+    }
+}
+
 }  // namespace paimon::test

src/paimon/core/mergetree/spill_channel_manager.h

@@ -0,0 +1,66 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <unordered_set>
+
+#include "paimon/disk/file_io_channel.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class SpillChannelManager {
+ public:
+    explicit SpillChannelManager(const std::shared_ptr<FileSystem>& fs,
+                                 size_t initial_capacity = 128)
+        : fs_(fs) {
+        channels_.reserve(initial_capacity);
+    }
+
+    void AddChannel(const std::shared_ptr<FileIOChannel::ID>& channel_id) {
+        channels_.emplace(*channel_id);
+    }
+
+    Status DeleteChannel(const std::shared_ptr<FileIOChannel::ID>& channel_id) {
+        PAIMON_RETURN_NOT_OK(fs_->Delete(channel_id->GetPath()));
+        channels_.erase(*channel_id);
+        return Status::OK();
+    }
+
+    Status Reset() {
+        Status first_error;
+        for (const auto& channel : channels_) {
+            auto status = fs_->Delete(channel.GetPath());
+            if (!status.ok() && first_error.ok()) {
+                first_error = std::move(status);
+            }
+        }
+        channels_.clear();
+        return first_error;
+    }
+
+    const std::unordered_set<FileIOChannel::ID, FileIOChannel::ID::Hash>& GetChannels() const {
+        return channels_;
+    }
+
+ private:
+    std::unordered_set<FileIOChannel::ID, FileIOChannel::ID::Hash> channels_;
+    std::shared_ptr<FileSystem> fs_;
+};
+
+}  // namespace paimon

src/paimon/core/mergetree/spill_channel_manager_test.cpp

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/mergetree/spill_channel_manager.h"
+
+#include <memory>
+#include <string>
+
+#include "gtest/gtest.h"
+#include "paimon/disk/io_manager.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class SpillChannelManagerTest : public ::testing::Test {
+ public:
+    void SetUp() override {
+        file_system_ = std::make_shared<LocalFileSystem>();
+        test_dir_ = UniqueTestDirectory::Create();
+        io_manager_ = IOManager::Create(test_dir_->Str());
+    }
+
+    std::shared_ptr<FileIOChannel::ID> CreateTempFile() {
+        auto result = io_manager_->CreateChannel();
+        EXPECT_TRUE(result.ok());
+        auto channel_id = result.value();
+        // Create the actual file on disk
+        auto out = file_system_->Create(channel_id->GetPath(), /*overwrite=*/false);
+        EXPECT_TRUE(out.ok());
+        EXPECT_OK(out.value()->Close());
+        return channel_id;
+    }
+
+ protected:
+    std::shared_ptr<LocalFileSystem> file_system_;
+    std::unique_ptr<UniqueTestDirectory> test_dir_;
+    std::shared_ptr<IOManager> io_manager_;
+};
+
+TEST_F(SpillChannelManagerTest, AddAndGetChannels) {
+    SpillChannelManager manager(file_system_);
+
+    auto channel1 = CreateTempFile();
+    auto channel2 = CreateTempFile();
+
+    manager.AddChannel(channel1);
+    manager.AddChannel(channel2);
+
+    const auto& channels = manager.GetChannels();
+    ASSERT_EQ(channels.size(), 2);
+    ASSERT_GT(channels.count(*channel1), 0);
+    ASSERT_GT(channels.count(*channel2), 0);
+}
+
+TEST_F(SpillChannelManagerTest, DeleteChannelRemovesFileAndEntry) {
+    SpillChannelManager manager(file_system_);
+
+    auto channel = CreateTempFile();
+    manager.AddChannel(channel);
+
+    ASSERT_OK_AND_ASSIGN(bool exists_before, file_system_->Exists(channel->GetPath()));
+    ASSERT_TRUE(exists_before);
+
+    ASSERT_OK(manager.DeleteChannel(channel));
+    ASSERT_EQ(manager.GetChannels().size(), 0);
+    ASSERT_OK_AND_ASSIGN(bool exists_after, file_system_->Exists(channel->GetPath()));
+    ASSERT_FALSE(exists_after);
+}
+
+TEST_F(SpillChannelManagerTest, ResetDeletesAllFiles) {
+    SpillChannelManager manager(file_system_);
+
+    auto channel1 = CreateTempFile();
+    auto channel2 = CreateTempFile();
+    auto channel3 = CreateTempFile();
+
+    manager.AddChannel(channel1);
+    manager.AddChannel(channel2);
+    manager.AddChannel(channel3);
+
+    ASSERT_OK_AND_ASSIGN(bool e1, file_system_->Exists(channel1->GetPath()));
+    ASSERT_OK_AND_ASSIGN(bool e2, file_system_->Exists(channel2->GetPath()));
+    ASSERT_OK_AND_ASSIGN(bool e3, file_system_->Exists(channel3->GetPath()));
+    ASSERT_TRUE(e1);
+    ASSERT_TRUE(e2);
+    ASSERT_TRUE(e3);
+
+    ASSERT_OK(manager.Reset());
+
+    ASSERT_OK_AND_ASSIGN(bool a1, file_system_->Exists(channel1->GetPath()));
+    ASSERT_OK_AND_ASSIGN(bool a2, file_system_->Exists(channel2->GetPath()));
+    ASSERT_OK_AND_ASSIGN(bool a3, file_system_->Exists(channel3->GetPath()));
+    ASSERT_FALSE(a1);
+    ASSERT_FALSE(a2);
+    ASSERT_FALSE(a3);
+}
+
+}  // namespace paimon::test