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

Author: dalingmeng

include/paimon/disk/file_io_channel.h

@@ -0,0 +1,72 @@
+/*
+ * 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 <memory>
+#include <random>
+#include <string>
+
+#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 {
+ public:
+    /// An ID identifying an underlying file channel.
+    class PAIMON_EXPORT ID {
+     public:
+        ID() = default;
+
+        explicit ID(const std::string& path);
+
+        ID(const std::string& base_path, std::mt19937* random);
+
+        ID(const std::string& base_path, const std::string& prefix, std::mt19937* random);
+
+        const std::string& GetPath() const;
+
+        bool operator==(const ID& other) const;
+
+        bool operator!=(const ID& other) const;
+
+        struct Hash {
+            size_t operator()(const ID& id) const;
+        };
+
+     private:
+        std::string path_;
+    };
+
+    /// An enumerator for channels that logically belong together.
+    class PAIMON_EXPORT Enumerator {
+     public:
+        Enumerator(const std::string& base_path, std::mt19937* random);
+
+        ID Next();
+
+     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<FileIOChannel::ID> CreateChannel() = 0;
+
+    virtual Result<FileIOChannel::ID> CreateChannel(const std::string& prefix) = 0;
+
+    virtual Result<std::shared_ptr<FileIOChannel::Enumerator>> CreateChannelEnumerator() = 0;
 };
 }  // namespace paimon

src/paimon/CMakeLists.txt

@@ -146,7 +146,10 @@ set(PAIMON_COMMON_SRCS
     common/utils/string_utils.cpp)
 
 set(PAIMON_CORE_SRCS
+    core/disk/file_io_channel.cpp
     core/disk/io_manager.cpp
+    core/mergetree/spill_reader.cpp
+    core/mergetree/spill_writer.cpp
     core/append/append_only_writer.cpp
     core/append/bucketed_append_compact_manager.cpp
     core/casting/binary_to_string_cast_executor.cpp
@@ -598,6 +601,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/file_io_channel.cpp

@@ -0,0 +1,80 @@
+/*
+ * 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/disk/file_io_channel.h"
+
+#include <iomanip>
+#include <sstream>
+#include <utility>
+
+#include "paimon/common/utils/path_util.h"
+
+namespace paimon {
+
+namespace {
+
+constexpr int32_t RANDOM_BYTES_LENGTH = 16;
+
+std::string GenerateRandomHexString(std::mt19937* random) {
+    std::uniform_int_distribution<int32_t> dist(0, 255);
+    std::ostringstream hex_stream;
+    hex_stream << std::hex << std::setfill('0');
+    for (int32_t i = 0; i < RANDOM_BYTES_LENGTH; ++i) {
+        hex_stream << std::setw(2) << dist(*random);
+    }
+    return hex_stream.str();
+}
+
+}  // namespace
+
+FileIOChannel::ID::ID(const std::string& path) : path_(path) {}
+
+FileIOChannel::ID::ID(const std::string& base_path, std::mt19937* random)
+    : path_(PathUtil::JoinPath(base_path, GenerateRandomHexString(random) + ".channel")) {}
+
+FileIOChannel::ID::ID(const std::string& base_path, const std::string& prefix, std::mt19937* random)
+    : path_(PathUtil::JoinPath(base_path,
+                               prefix + "-" + GenerateRandomHexString(random) + ".channel")) {}
+
+const std::string& FileIOChannel::ID::GetPath() const {
+    return path_;
+}
+
+bool FileIOChannel::ID::operator==(const ID& other) const {
+    return path_ == other.path_;
+}
+
+bool FileIOChannel::ID::operator!=(const ID& other) const {
+    return !(*this == other);
+}
+
+size_t FileIOChannel::ID::Hash::operator()(const ID& id) const {
+    return std::hash<std::string>{}(id.path_);
+}
+
+FileIOChannel::Enumerator::Enumerator(const std::string& base_path, std::mt19937* random)
+    : path_(base_path), name_prefix_(GenerateRandomHexString(random)) {}
+
+FileIOChannel::ID FileIOChannel::Enumerator::Next() {
+    std::ostringstream filename;
+    filename << name_prefix_ << "." << std::setfill('0') << std::setw(6) << (local_counter_++)
+             << ".channel";
+
+    std::string full_path = PathUtil::JoinPath(path_, filename.str());
+    return ID(full_path);
+}
+
+}  // namespace paimon

src/paimon/core/disk/io_manager.cpp

@@ -15,13 +15,18 @@
  */
 #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) {
+        std::random_device rd;
+        random_.seed(rd());
+    }
 
     const std::string& GetTempDir() const override {
         return tmp_dir_;
@@ -35,9 +40,26 @@ class IOManagerImpl : public IOManager {
         return PathUtil::JoinPath(tmp_dir_, prefix + "-" + uuid + std::string(kSuffix));
     }
 
+    Result<FileIOChannel::ID> CreateChannel() override {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return FileIOChannel::ID(tmp_dir_, &random_);
+    }
+
+    Result<FileIOChannel::ID> CreateChannel(const std::string& prefix) override {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return 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,59 @@
+/*
+ * 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:
+    SpillChannelManager(const std::shared_ptr<FileSystem>& fs, size_t initial_capacity) : fs_(fs) {
+        channels_.reserve(initial_capacity);
+    }
+
+    void AddChannel(const FileIOChannel::ID& channel_id) {
+        channels_.emplace(channel_id);
+    }
+
+    Status DeleteChannel(const FileIOChannel::ID& channel_id) {
+        channels_.erase(channel_id);
+        PAIMON_RETURN_NOT_OK(fs_->Delete(channel_id.GetPath()));
+        return Status::OK();
+    }
+
+    void Reset() {
+        for (const auto& channel : channels_) {
+            fs_->Delete(channel.GetPath());
+        }
+        channels_.clear();
+    }
+
+    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