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,8 @@ 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_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/common/utils/arrow/arrow_utils_test.cpp

@@ -445,4 +445,43 @@ TEST(ArrowUtilsTest, TestEqualsIgnoreNullable) {
         ASSERT_TRUE(ArrowUtils::EqualsIgnoreNullable(struct_type1, struct_type2));
     }
 }
+
+TEST(ArrowUtilsTest, TestGetCompressionType) {
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType(""));
+        ASSERT_EQ(type, arrow::Compression::UNCOMPRESSED);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("none"));
+        ASSERT_EQ(type, arrow::Compression::UNCOMPRESSED);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("uncompressed"));
+        ASSERT_EQ(type, arrow::Compression::UNCOMPRESSED);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("zstd"));
+        ASSERT_EQ(type, arrow::Compression::ZSTD);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("ZSTD"));
+        ASSERT_EQ(type, arrow::Compression::ZSTD);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("lz4"));
+        ASSERT_EQ(type, arrow::Compression::LZ4_FRAME);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("snappy"));
+        ASSERT_EQ(type, arrow::Compression::SNAPPY);
+    }
+    {
+        ASSERT_OK_AND_ASSIGN(auto type, ArrowUtils::GetCompressionType("gzip"));
+        ASSERT_EQ(type, arrow::Compression::GZIP);
+    }
+    {
+        ASSERT_NOK(ArrowUtils::GetCompressionType("invalid_codec"));
+    }
+}
+
 }  // namespace paimon::test

src/paimon/core/disk/file_channel_manager.h

@@ -0,0 +1,91 @@
+/*
+ * 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 <memory>
+#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/fs/local/local_file_system.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+/// Manages the creation and lifecycle of spill file channels.
+class FileChannelManager {
+ public:
+    static Result<std::unique_ptr<FileChannelManager>> Create(const std::string& tmp_dir,
+                                                              const std::string& prefix) {
+        std::string uuid;
+        if (!UUID::Generate(&uuid)) {
+            return Status::Invalid("Failed to generate UUID for FileChannelManager.");
+        }
+        std::string spill_dir = PathUtil::JoinPath(tmp_dir, "paimon-" + prefix + "-" + uuid);
+
+        auto fs = std::make_shared<LocalFileSystem>();
+        PAIMON_RETURN_NOT_OK(fs->Mkdirs(spill_dir));
+
+        std::random_device rd;
+        std::mt19937 random(rd());
+
+        return std::unique_ptr<FileChannelManager>(
+            new FileChannelManager(std::move(spill_dir), std::move(random), std::move(fs)));
+    }
+
+    ~FileChannelManager() {
+        if (!spill_dir_.empty() && fs_ != nullptr) {
+            [[maybe_unused]] auto status = fs_->Delete(spill_dir_, /*recursive=*/true);
+        }
+    }
+
+    FileChannelManager(const FileChannelManager&) = delete;
+    FileChannelManager& operator=(const FileChannelManager&) = delete;
+
+    FileIOChannel::ID CreateChannel() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return FileIOChannel::ID(spill_dir_, &random_);
+    }
+
+    FileIOChannel::ID CreateChannel(const std::string& prefix) {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return FileIOChannel::ID(spill_dir_, prefix, &random_);
+    }
+
+    std::shared_ptr<FileIOChannel::Enumerator> CreateChannelEnumerator() {
+        std::lock_guard<std::mutex> lock(mutex_);
+        return std::make_shared<FileIOChannel::Enumerator>(spill_dir_, &random_);
+    }
+
+    const std::string& GetSpillDir() const {
+        return spill_dir_;
+    }
+
+ private:
+    FileChannelManager(const std::string& spill_dir, std::mt19937&& random,
+                       const std::shared_ptr<FileSystem>& fs)
+        : spill_dir_(spill_dir), random_(std::move(random)), fs_(fs) {}
+    std::string spill_dir_;
+    std::mutex mutex_;
+    std::mt19937 random_;
+    std::shared_ptr<FileSystem> fs_;
+};
+
+}  // namespace paimon

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);