feat(file-index): support writing file indexes

include/paimon/defs.h

@@ -405,6 +405,10 @@ struct PAIMON_EXPORT Options {
     /// "file-index.read.enabled" - Whether enabled read file index. Default value is "true".
     static const char FILE_INDEX_READ_ENABLED[];
 
+    /// "file-index.in-manifest-threshold" - The threshold to store file index bytes in the
+    /// manifest. Default value is 500B.
+    static const char FILE_INDEX_IN_MANIFEST_THRESHOLD[];
+
     /// "data-file.external-paths" - The external paths where the data of this table will be
     /// written, multiple elements separated by commas.
     static const char DATA_FILE_EXTERNAL_PATHS[];

include/paimon/file_index/file_index_format.h

@@ -19,6 +19,7 @@
 #pragma once
 
 #include <cstdint>
+#include <map>
 #include <memory>
 #include <string>
 #include <vector>
@@ -32,6 +33,8 @@ struct ArrowSchema;
 namespace paimon {
 class InputStream;
 class MemoryPool;
+class Bytes;
+class OutputStream;
 
 /// Defines the on-disk format and versioning for Paimon file-level indexes.
 /// File index file format. Put all column and offset in the header.
@@ -88,28 +91,54 @@ class MemoryPool;
 class PAIMON_EXPORT FileIndexFormat {
  public:
     class Reader;
+    class Writer;
+
+    /// Serialized file indexes grouped as column name -> index type -> index bytes. A null bytes
+    /// pointer represents an empty index for that column and index type.
+    /// For example, indexes["col1"]["bsi"] = <bytes>;
+    using ColumnIndexes = std::map<std::string, std::map<std::string, std::shared_ptr<Bytes>>>;
+
     /// Creates a `Reader` to parse a index file (may contain multiple indexes) from the given input
     /// stream.
-    ///
     /// @param input_stream Input stream containing serialized index data.
     /// @param pool Memory pool for temporary allocations during reading.
     /// @return A unique pointer to a `Reader` on success, or an error if the stream is invalid
     ///         (e.g., wrong magic, unsupported version, or corrupted data).
     static Result<std::unique_ptr<Reader>> CreateReader(
         const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<MemoryPool>& pool);
 
+    /// Creates a `Writer` which serializes a complete V1 file index container.
+    ///
+    /// @param output_stream Destination stream for serialized index data.
+    /// @param pool Memory pool for writer-side allocations.
+    /// @return A unique pointer to a `Writer` on success.
+    static Result<std::unique_ptr<Writer>> CreateWriter(
+        const std::shared_ptr<OutputStream>& output_stream,
+        const std::shared_ptr<MemoryPool>& pool);
+
  public:
     static const int64_t MAGIC;
     static const int32_t EMPTY_INDEX_FLAG;
     static const int32_t V_1;
 };
 
+/// Writer for file index file.
+class FileIndexFormat::Writer {
+ public:
+    virtual ~Writer() = default;
+
+    /// Writes all column indexes. This is a terminal, one-shot operation.
+    virtual Status WriteColumnIndexes(const FileIndexFormat::ColumnIndexes& indexes) = 0;
+
+    /// Flushes and closes the output stream supplied to `CreateWriter()`.
+    virtual Status Close() = 0;
+};
+
 /// Reader for file index file.
 class FileIndexFormat::Reader {
  public:
     virtual ~Reader() = default;
     /// Reads index data for a specific column from the index file.
-    ///
     /// @param column_name Name of the column to retrieve index data for.
     /// @param arrow_schema Arrow schema that must contain a field corresponding to `column_name`.
     /// @return A vector of shared pointers to FileIndexReader objects, each corresponding to a

src/paimon/CMakeLists.txt

@@ -85,6 +85,7 @@ set(PAIMON_COMMON_SRCS
     common/global_index/global_indexer_factory.cpp
     common/io/buffered_input_stream.cpp
     common/io/byte_array_input_stream.cpp
+    common/io/byte_array_output_stream.cpp
     common/io/data_input_stream.cpp
     common/io/data_output_stream.cpp
     common/io/memory_segment_output_stream.cpp
@@ -270,6 +271,8 @@ set(PAIMON_CORE_SRCS
     core/io/data_file_meta.cpp
     core/io/data_file_meta_serializer.cpp
     core/io/data_file_path_factory.cpp
+    core/io/data_file_index_writer.cpp
+    core/io/file_index_options.cpp
     core/io/append_data_file_writer_factory.cpp
     core/io/blob_data_file_writer_factory.cpp
     core/io/data_file_writer_factory.cpp
@@ -577,6 +580,7 @@ if(PAIMON_BUILD_TESTS)
                     common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
                     common/global_index/wrap/file_index_reader_wrapper_test.cpp
                     common/io/byte_array_input_stream_test.cpp
+                    common/io/byte_array_output_stream_test.cpp
                     common/io/data_input_output_stream_test.cpp
                     common/io/buffered_input_stream_test.cpp
                     common/io/memory_segment_output_stream_test.cpp
@@ -752,6 +756,8 @@ if(PAIMON_BUILD_TESTS)
                     core/io/complete_row_tracking_fields_reader_test.cpp
                     core/io/vector_file_batch_reader_test.cpp
                     core/io/data_file_meta_test.cpp
+                    core/io/data_file_index_writer_test.cpp
+                    core/io/file_index_options_test.cpp
                     core/io/file_index_evaluator_test.cpp
                     core/io/single_file_writer_test.cpp
                     core/io/rolling_blob_file_writer_test.cpp

src/paimon/common/defs.cpp

@@ -100,6 +100,7 @@ const char Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP[] =
 const char Options::SCAN_FALLBACK_BRANCH[] = "scan.fallback-branch";
 const char Options::BRANCH[] = "branch";
 const char Options::FILE_INDEX_READ_ENABLED[] = "file-index.read.enabled";
+const char Options::FILE_INDEX_IN_MANIFEST_THRESHOLD[] = "file-index.in-manifest-threshold";
 const char Options::DATA_FILE_EXTERNAL_PATHS[] = "data-file.external-paths";
 const char Options::DATA_FILE_EXTERNAL_PATHS_STRATEGY[] = "data-file.external-paths.strategy";
 const char Options::DATA_FILE_PREFIX[] = "data-file.prefix";

src/paimon/common/file_index/file_index_format.cpp

@@ -27,7 +27,9 @@
 #include "arrow/type.h"
 #include "fmt/format.h"
 #include "paimon/common/file_index/empty/empty_file_index_reader.h"
+#include "paimon/common/io/data_output_stream.h"
 #include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/math.h"
 #include "paimon/file_index/file_indexer.h"
 #include "paimon/file_index/file_indexer_factory.h"
 #include "paimon/io/byte_array_input_stream.h"
@@ -39,6 +41,128 @@ namespace paimon {
 class InputStream;
 class MemoryPool;
 
+class FileIndexFormatWriterImpl : public FileIndexFormat::Writer {
+ public:
+    explicit FileIndexFormatWriterImpl(const std::shared_ptr<OutputStream>& output_stream)
+        : output_stream_(output_stream) {
+        assert(output_stream_);
+    }
+
+    Status WriteColumnIndexes(const FileIndexFormat::ColumnIndexes& indexes) override {
+        if (written_) {
+            return Status::Invalid("File index column indexes have already been written");
+        }
+
+        PAIMON_RETURN_NOT_OK(WriteHead(indexes));
+        // Write body.
+        DataOutputStream data_output(output_stream_);
+        for (const auto& [column_name, column_indexes] : indexes) {
+            for (const auto& [index_type, bytes] : column_indexes) {
+                if (bytes) {
+                    PAIMON_RETURN_NOT_OK(data_output.WriteBytes(bytes));
+                }
+            }
+        }
+        written_ = true;
+        return Status::OK();
+    }
+
+    Status Close() override {
+        if (closed_) {
+            return Status::OK();
+        }
+        closed_ = true;
+        PAIMON_RETURN_NOT_OK(output_stream_->Flush());
+        return output_stream_->Close();
+    }
+
+ private:
+    static constexpr int32_t kRedundantLength = 0;
+
+    static Result<int32_t> CalculateHeadLength(const FileIndexFormat::ColumnIndexes& indexes) {
+        // magic(8), version(4), header length(4), and column count(4).
+        int64_t head_length = 8 + 4 + 4 + 4;
+        int64_t body_length = 0;
+        PAIMON_RETURN_NOT_OK(
+            ValidateValueInRange<int32_t>(indexes.size(), "file index column count"));
+        for (const auto& [column_name, column_indexes] : indexes) {
+            PAIMON_RETURN_NOT_OK(ValidateValueInRange<uint16_t>(column_name.size(),
+                                                                "file index column name length"));
+            PAIMON_RETURN_NOT_OK(
+                ValidateValueInRange<int32_t>(column_indexes.size(), "column index count"));
+            // column name(2 + N) + index count(4)
+            head_length += 2 + static_cast<int64_t>(column_name.size()) + 4;
+            for (const auto& [index_type, bytes] : column_indexes) {
+                PAIMON_RETURN_NOT_OK(ValidateValueInRange<uint16_t>(index_type.size(),
+                                                                    "file index type name length"));
+                // index type(2 + N) + body offset(4) + body length(4)
+                head_length += 2 + static_cast<int64_t>(index_type.size()) + 4 + 4;
+                if (bytes) {
+                    PAIMON_RETURN_NOT_OK(AddChecked(bytes->size(), "index body", &body_length));
+                }
+            }
+        }
+
+        head_length += 4;  // The trailing redundant-length field(4).
+        PAIMON_RETURN_NOT_OK(
+            ValidateValueInRange<int32_t>(head_length, "file index header length"));
+        int64_t container_length = head_length + body_length;
+        PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(container_length, "file index size"));
+        return static_cast<int32_t>(head_length);
+    }
+
+    Status WriteHead(const FileIndexFormat::ColumnIndexes& indexes) {
+        PAIMON_ASSIGN_OR_RAISE(int32_t head_length, CalculateHeadLength(indexes));
+        DataOutputStream data_output(output_stream_);
+        // Write magic.
+        PAIMON_RETURN_NOT_OK(data_output.WriteValue<int64_t>(FileIndexFormat::MAGIC));
+        // Write version.
+        PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(FileIndexFormat::V_1));
+        // Write head length.
+        PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(head_length));
+        // Write column count.
+        PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(static_cast<int32_t>(indexes.size())));
+
+        int64_t body_offset = head_length;
+        for (const auto& [column_name, column_indexes] : indexes) {
+            // Write column name.
+            PAIMON_RETURN_NOT_OK(data_output.WriteString(column_name));
+            // Write index count for the column.
+            PAIMON_RETURN_NOT_OK(
+                data_output.WriteValue<int32_t>(static_cast<int32_t>(column_indexes.size())));
+            for (const auto& [index_type, bytes] : column_indexes) {
+                // Write index type.
+                PAIMON_RETURN_NOT_OK(data_output.WriteString(index_type));
+                // Write body offset and length.
+                if (bytes) {
+                    PAIMON_RETURN_NOT_OK(
+                        data_output.WriteValue<int32_t>(static_cast<int32_t>(body_offset)));
+                    PAIMON_RETURN_NOT_OK(
+                        data_output.WriteValue<int32_t>(static_cast<int32_t>(bytes->size())));
+                    body_offset += static_cast<int64_t>(bytes->size());
+                } else {
+                    PAIMON_RETURN_NOT_OK(
+                        data_output.WriteValue<int32_t>(FileIndexFormat::EMPTY_INDEX_FLAG));
+                    PAIMON_RETURN_NOT_OK(data_output.WriteValue<int32_t>(0));
+                }
+            }
+        }
+        // Write redundant length for future format extensions.
+        return data_output.WriteValue<int32_t>(kRedundantLength);
+    }
+
+    template <typename T>
+    static Status AddChecked(T value, const char* name, int64_t* total) {
+        PAIMON_RETURN_NOT_OK(ValidateValueInRange<int32_t>(value, name));
+        *total += static_cast<int64_t>(value);
+        return ValidateValueInRange<int32_t>(*total, name);
+    }
+
+    std::shared_ptr<OutputStream> output_stream_;
+    bool written_ = false;
+    bool closed_ = false;
+};
+
 class FileIndexFormatReaderImpl : public FileIndexFormat::Reader {
  public:
     using HeaderType =
@@ -153,4 +277,15 @@ Result<std::unique_ptr<FileIndexFormat::Reader>> FileIndexFormat::CreateReader(
     const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<MemoryPool>& pool) {
     return FileIndexFormatReaderImpl::Create(input_stream, pool);
 }
+
+Result<std::unique_ptr<FileIndexFormat::Writer>> FileIndexFormat::CreateWriter(
+    const std::shared_ptr<OutputStream>& output_stream, const std::shared_ptr<MemoryPool>& pool) {
+    if (!output_stream) {
+        return Status::Invalid("File index output stream cannot be null");
+    }
+    if (!pool) {
+        return Status::Invalid("File index memory pool cannot be null");
+    }
+    return std::make_unique<FileIndexFormatWriterImpl>(output_stream);
+}
 }  // namespace paimon

src/paimon/common/file_index/file_index_format_test.cpp

@@ -24,17 +24,20 @@
 #include "paimon/common/file_index/bloomfilter/bloom_filter_file_index.h"
 #include "paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h"
 #include "paimon/common/file_index/empty/empty_file_index_reader.h"
+#include "paimon/common/io/byte_array_output_stream.h"
 #include "paimon/data/timestamp.h"
 #include "paimon/defs.h"
 #include "paimon/file_index/file_index_result.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/io/byte_array_input_stream.h"
+#include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/predicate/literal.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
+
 class FileIndexFormatTest : public ::testing::Test {
  public:
     void SetUp() override {
@@ -55,14 +58,25 @@ class FileIndexFormatTest : public ::testing::Test {
     std::shared_ptr<MemoryPool> pool_;
 };
 
-TEST_F(FileIndexFormatTest, TestCreateEmptyFileIndexReader) {
+TEST_F(FileIndexFormatTest, TestWriteAndReadEmptyIndexGoldenBytes) {
+    // the expected bytes are generated from Java Paimon
+    std::vector<char> expected = {0,   5,   78,  78, -48, 26, 53, -82, 0, 0, 0, 1, 0, 0, 0,   47,
+                                  0,   0,   0,   1,  0,   2,  99, 49,  0, 0, 0, 1, 0, 5, 101, 109,
+                                  112, 116, 121, -1, -1,  -1, -1, 0,   0, 0, 0, 0, 0, 0, 0};
+    FileIndexFormat::ColumnIndexes indexes;
+    indexes["c1"]["empty"] = nullptr;
+    auto segment_output = std::make_unique<MemorySegmentOutputStream>(
+        MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool_);
+    auto output = std::make_shared<ByteArrayOutputStream>(std::move(segment_output));
+
+    ASSERT_OK_AND_ASSIGN(auto writer, FileIndexFormat::CreateWriter(output, pool_));
+    ASSERT_OK(writer->WriteColumnIndexes(indexes));
+    ASSERT_OK(writer->Close());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> actual, output->Finish(pool_.get()));
+
+    ASSERT_EQ(expected, std::vector<char>(actual->data(), actual->data() + actual->size()));
     auto schema = arrow::schema({arrow::field("c1", arrow::utf8())});
-    std::vector<char> index_file_bytes = {0,  5,  78, 78, -48, 26, 53,  -82, 0,   0,   0,   1,
-                                          0,  0,  0,  47, 0,   0,  0,   1,   0,   2,   99,  49,
-                                          0,  0,  0,  1,  0,   5,  101, 109, 112, 116, 121, -1,
-                                          -1, -1, -1, 0,  0,   0,  0,   0,   0,   0,   0};
-    auto input_stream =
-        std::make_shared<ByteArrayInputStream>(index_file_bytes.data(), index_file_bytes.size());
+    auto input_stream = std::make_shared<ByteArrayInputStream>(actual->data(), actual->size());
     ASSERT_OK_AND_ASSIGN(auto reader, FileIndexFormat::CreateReader(input_stream, pool_));
     ASSERT_OK_AND_ASSIGN(auto index_file_readers,
                          reader->ReadColumnIndex("c1", CreateArrowSchema(schema).get()));