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

Author: dalingmeng

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
@@ -605,6 +608,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/common/utils/arrow/arrow_utils.cpp

@@ -18,7 +18,9 @@
 
 #include "arrow/array/array_base.h"
 #include "arrow/array/array_nested.h"
+#include "arrow/util/compression.h"
 #include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/string_utils.h"
 
 namespace paimon {
 Result<std::shared_ptr<arrow::Schema>> ArrowUtils::DataTypeToSchema(
@@ -161,4 +163,14 @@ Result<std::shared_ptr<arrow::StructArray>> ArrowUtils::RemoveFieldFromStructArr
     return array;
 }
 
+Result<arrow::Compression::type> ArrowUtils::GetCompressionType(const std::string& compression) {
+    std::string normalized = StringUtils::ToLowerCase(compression);
+    if (normalized.empty() || normalized == "none") {
+        normalized = "uncompressed";
+    }
+    PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(arrow::Compression::type compression_type,
+                                      arrow::util::Codec::GetCompressionType(normalized));
+    return compression_type;
+}
+
 }  // namespace paimon

src/paimon/common/utils/arrow/arrow_utils.h

@@ -19,6 +19,7 @@
 #include <vector>
 
 #include "arrow/api.h"
+#include "arrow/util/type_fwd.h"
 #include "fmt/format.h"
 #include "paimon/result.h"
 
@@ -49,6 +50,10 @@ class PAIMON_EXPORT ArrowUtils {
     static bool EqualsIgnoreNullable(const std::shared_ptr<arrow::DataType>& type,
                                      const std::shared_ptr<arrow::DataType>& other_type);
 
+    /// Normalize and resolve a compression string to an Arrow compression type.
+    /// Handles "none" and empty string by mapping them to "uncompressed".
+    static Result<arrow::Compression::type> GetCompressionType(const std::string& compression);
+
  private:
     static Status InnerCheckNullabilityMatch(const std::shared_ptr<arrow::Field>& field,
                                              const std::shared_ptr<arrow::Array>& data);

src/paimon/core/disk/file_io_channel.cpp

@@ -0,0 +1,74 @@
+/*
+ * 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/disk/file_io_channel.h"
+
+#include <iomanip>
+#include <sstream>
+#include <utility>
+
+#include "paimon/common/utils/path_util.h"
+
+namespace paimon {
+
+std::string FileIOChannel::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 < kRandomBytesLength; ++i) {
+        hex_stream << std::setw(2) << dist(*random);
+    }
+    return hex_stream.str();
+}
+
+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/file_io_channel.h

@@ -0,0 +1,73 @@
+/*
+ * 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.
+class PAIMON_EXPORT FileIOChannel {
+ public:
+    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_;
+    };
+
+ private:
+    static constexpr int32_t kRandomBytesLength = 16;
+    static std::string GenerateRandomHexString(std::mt19937* random);
+
+ public:
+    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

src/paimon/core/disk/io_manager.cpp

@@ -13,32 +13,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-#include "paimon/disk/io_manager.h"
-
-#include "paimon/common/utils/path_util.h"
-#include "paimon/common/utils/uuid.h"
+#include "paimon/core/disk/io_manager_impl.h"
 
 namespace paimon {
-class IOManagerImpl : public IOManager {
- public:
-    explicit IOManagerImpl(const std::string& tmp_dir) : tmp_dir_(tmp_dir) {}
-
-    const std::string& GetTempDir() const override {
-        return tmp_dir_;
-    }
-
-    Result<std::string> GenerateTempFilePath(const std::string& prefix) const override {
-        std::string uuid;
-        if (!UUID::Generate(&uuid)) {
-            return Status::Invalid("generate uuid for io manager tmp path failed.");
-        }
-        return PathUtil::JoinPath(tmp_dir_, prefix + "-" + uuid + std::string(kSuffix));
-    }
-
- private:
-    static constexpr char kSuffix[] = ".channel";
-    std::string tmp_dir_;
-};
 
 std::unique_ptr<IOManager> IOManager::Create(const std::string& tmp_dir) {
     return std::make_unique<IOManagerImpl>(tmp_dir);

src/paimon/core/disk/io_manager_impl.h

@@ -0,0 +1,71 @@
+/*
+ * 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 <mutex>
+#include <random>
+#include <string>
+
+#include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/uuid.h"
+#include "paimon/core/disk/file_io_channel.h"
+#include "paimon/disk/io_manager.h"
+
+namespace paimon {
+
+class IOManagerImpl : public IOManager {
+ public:
+    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_;
+    }
+
+    Result<std::string> GenerateTempFilePath(const std::string& prefix) const override {
+        std::string uuid;
+        if (!UUID::Generate(&uuid)) {
+            return Status::Invalid("generate uuid for io manager tmp path failed.");
+        }
+        return PathUtil::JoinPath(tmp_dir_, prefix + "-" + uuid + std::string(kSuffix));
+    }
+
+    Result<FileIOChannel::ID> CreateChannel() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return FileIOChannel::ID(tmp_dir_, &random_);
+    }
+
+    Result<FileIOChannel::ID> CreateChannel(const std::string& prefix) {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return FileIOChannel::ID(tmp_dir_, prefix, &random_);
+    }
+
+    Result<std::shared_ptr<FileIOChannel::Enumerator>> CreateChannelEnumerator() {
+        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_;
+};
+
+}  // namespace paimon

src/paimon/core/disk/io_manager_test.cpp

@@ -14,14 +14,13 @@
  * limitations under the License.
  */
 
-#include "paimon/disk/io_manager.h"
-
 #include <memory>
 #include <string>
 
 #include "gtest/gtest.h"
 #include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/string_utils.h"
+#include "paimon/core/disk/io_manager_impl.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
@@ -63,4 +62,33 @@ TEST(IOManagerTest, GenerateTempFilePathShouldBeDifferentAcrossCalls) {
     ASSERT_NE(path1, path2);
 }
 
+TEST(IOManagerTest, CreateChannelShouldReturnValidAndUniquePaths) {
+    auto tmp_dir = UniqueTestDirectory::Create();
+    auto manager = std::make_unique<IOManagerImpl>(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();
+    auto manager = std::make_unique<IOManagerImpl>(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