feat(file-index): support V2 containers and streaming writes

include/paimon/defs.h

@@ -469,6 +469,9 @@ struct PAIMON_EXPORT Options {
     /// manifest. Default value is 500B.
     static const char FILE_INDEX_IN_MANIFEST_THRESHOLD[];
 
+    /// "file-index.format.version" - File index container write version, 1 or 2. Defaults to 1.
+    static const char FILE_INDEX_FORMAT_VERSION[];
+
     /// "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 <functional>
 #include <map>
 #include <memory>
 #include <string>
@@ -37,57 +38,10 @@ 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.
-///
-/// <pre>
-///  _______________________________________    _____________________
-/// |     magic    | version | head length |
-/// |--------------------------------------|
-/// |            column number             |
-/// |--------------------------------------|
-/// |   column 1        |  index number    |
-/// |--------------------------------------|
-/// |  index name 1 | start pos | length   |
-/// |--------------------------------------|
-/// |  index name 2 | start pos | length   |
-/// |--------------------------------------|
-/// |  index name 3 | start pos | length   |
-/// |--------------------------------------|            HEAD
-/// |   column 2        |  index number    |
-/// |--------------------------------------|
-/// |  index name 1 | start pos | length   |
-/// |--------------------------------------|
-/// |  index name 2 | start pos | length   |
-/// |--------------------------------------|
-/// |  index name 3 | start pos | length   |
-/// |--------------------------------------|
-/// |                 ...                  |
-/// |--------------------------------------|
-/// |                 ...                  |
-/// |--------------------------------------|
-/// |  redundant length | redundant bytes  |
-/// |--------------------------------------|    ---------------------
-/// |                BODY                  |
-/// |                BODY                  |
-/// |                BODY                  |             BODY
-/// |                BODY                  |
-/// |______________________________________|    _____________________
-///
-/// magic:               8 bytes long
-/// version:             4 bytes int
-/// head length:         4 bytes int
-/// column number:       4 bytes int
-/// column x:            var bytes utf (length + bytes)
-/// index number:        4 bytes int (how many column items below)
-/// index name x:        var bytes utf
-/// start pos:           4 bytes int
-/// length:              4 bytes int
-/// redundant length:    4 bytes int (for compatibility with later versions, in this version,
-///                      content is zero)
-/// redundant bytes:     var bytes (for compatibility with later version, in this version, is empty)
-/// BODY:                column index bytes + column index bytes + column index bytes + .......
-/// </pre>
-///
+/// Supports V1 and V2 containers. V2 uses int64 offsets and lengths and supports streaming index
+/// payloads directly to the output; avoiding full payload buffering requires a streaming WriteTo()
+/// implementation in the index plugin.
+/// See file_index_format_v1.h and file_index_format_v2.h for the serialized layouts.
 class PAIMON_EXPORT FileIndexFormat {
  public:
     class Reader;
@@ -109,35 +63,49 @@ class PAIMON_EXPORT FileIndexFormat {
         const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<MemoryPool>& pool,
         const std::map<std::string, std::string>& options);
 
-    /// Creates a `Writer` which serializes a complete V1 file index container.
+    /// Creates a `Writer` for file index container.
     ///
     /// @param output_stream Destination stream for serialized index data.
+    /// @param version Container format version (V1 or V2).
     /// @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<OutputStream>& output_stream, int32_t version,
         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;
+    static const int32_t V_2;
 };
 
 /// Writer for file index file.
-class FileIndexFormat::Writer {
+class PAIMON_EXPORT FileIndexFormat::Writer {
  public:
+    /// Writes one payload without closing the stream. Returns false for an absent index
+    /// (without writing bytes), or true after writing a non-empty payload.
+    /// The borrowed stream reports positions relative to the start of this payload, starting at 0.
+    /// The callback must not retain the stream.
+    using PayloadWriter = std::function<Result<bool>(OutputStream*)>;
     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;
+
+    /// Writes one index. V1 buffers payloads; V2 writes them directly to the destination.
+    /// An empty callback or a false result denotes an empty index (-1, 0).
+    /// A true result indicates that a non-empty payload was written.
+    /// Call Finish() after the last index. Payload failures are terminal.
+    virtual Status WriteIndex(const std::string& column_name, const std::string& index_type,
+                              const PayloadWriter& payload_writer) = 0;
+
+    /// Completes the container: writes V1's header and buffered bodies, or V2's footer and trailer.
+    virtual Status Finish() = 0;
 };
 
 /// Reader for file index file.
-class FileIndexFormat::Reader {
+class PAIMON_EXPORT FileIndexFormat::Reader {
  public:
     virtual ~Reader() = default;
     /// Reads index data for a specific column from the index file.

include/paimon/file_index/file_index_writer.h

@@ -29,6 +29,8 @@
 struct ArrowArray;
 
 namespace paimon {
+class OutputStream;
+
 /// Interface for writing file-level index data from Arrow batches.
 class PAIMON_EXPORT FileIndexWriter {
  public:
@@ -48,9 +50,22 @@ class PAIMON_EXPORT FileIndexWriter {
     /// have been added. It can be called only once and typically assumes no further calls to
     /// `AddBatch()` will occur afterward.
     ///
+    /// Streaming-only writers may override WriteTo() and use the default NotImplemented result
+    /// from this method.
+    ///
     /// @return A unique pointer to a byte array containing the serialized index data,
     ///         or an error if serialization fails.
-    virtual Result<PAIMON_UNIQUE_PTR<Bytes>> SerializedBytes() const = 0;
+    virtual Result<PAIMON_UNIQUE_PTR<Bytes>> SerializedBytes() const;
+
+    /// Writes one index payload to the caller-owned stream without closing it.
+    /// The stream is borrowed for this call and must not be retained by the writer.
+    /// The supplied stream must report positions relative to the start of this index payload.
+    /// The default implementation writes SerializedBytes(); streaming writers can override this
+    /// method to avoid materializing the complete payload. Write failures must be propagated.
+    /// @return true if a non-empty index payload was produced, false if there is no index, or an
+    /// error. Returning false must not write any bytes. The default implementation treats null or
+    /// zero-length SerializedBytes() results as no index.
+    virtual Result<bool> WriteTo(OutputStream* output) const;
 };
 
 }  // namespace paimon

include/paimon/file_index/file_indexer.h

@@ -44,6 +44,7 @@ class PAIMON_EXPORT FileIndexer {
     }
 
     /// Create `FileIndexReader` with input stream.
+    /// @note Plugin implementations should validate their own payload size limits.
     ///
     /// @param arrow_schema ArrowSchema derived from arrow schema or struct type with
     /// specified indexed field.
@@ -53,7 +54,7 @@ class PAIMON_EXPORT FileIndexer {
     /// @param pool Memory pool for memory allocation.
     /// @return A `FileIndexReader` to read index.
     virtual Result<std::shared_ptr<FileIndexReader>> CreateReader(
-        ::ArrowSchema* arrow_schema, int32_t start, int32_t length,
+        ::ArrowSchema* arrow_schema, int64_t start, int64_t length,
         const std::shared_ptr<InputStream>& input_stream,
         const std::shared_ptr<MemoryPool>& pool) const = 0;
 

src/paimon/CMakeLists.txt

@@ -71,7 +71,11 @@ set(PAIMON_COMMON_SRCS
     common/file_index/bitmap/bitmap_index_result.cpp
     common/file_index/file_indexer_factory.cpp
     common/file_index/file_index_format.cpp
+    common/file_index/file_index_format_utils.cpp
+    common/file_index/file_index_format_v1.cpp
+    common/file_index/file_index_format_v2.cpp
     common/file_index/file_index_reader.cpp
+    common/file_index/file_index_writer.cpp
     common/file_index/file_index_result.cpp
     common/file_index/scored_file_index_result.cpp
     common/format/column_stats.cpp
@@ -93,6 +97,7 @@ set(PAIMON_COMMON_SRCS
     common/io/data_output_stream.cpp
     common/io/memory_segment_output_stream.cpp
     common/io/offset_input_stream.cpp
+    common/io/offset_output_stream.cpp
     common/io/cache/cache.cpp
     common/io/cache/cache_key.cpp
     common/io/cache/cache_manager.cpp
@@ -294,6 +299,7 @@ set(PAIMON_CORE_SRCS
     core/io/data_file_meta_write_cols_legacy_serializer.cpp
     core/io/data_file_path_factory.cpp
     core/io/data_file_index_writer.cpp
+    core/io/spillable_index_output_stream.cpp
     core/io/file_index_options.cpp
     core/io/append_data_file_writer_factory.cpp
     core/io/blob_data_file_writer_factory.cpp
@@ -597,6 +603,7 @@ if(PAIMON_BUILD_TESTS)
                     common/file_index/file_indexer_factory_test.cpp
                     common/file_index/file_index_result_test.cpp
                     common/file_index/file_index_reader_test.cpp
+                    common/file_index/file_index_writer_test.cpp
                     common/file_index/scored_file_index_result_test.cpp
                     common/file_index/file_index_format_test.cpp
                     common/file_index/empty/empty_file_index_reader_test.cpp
@@ -639,6 +646,7 @@ if(PAIMON_BUILD_TESTS)
                     common/io/memory_segment_output_stream_test.cpp
                     common/io/cache_input_stream_test.cpp
                     common/io/offset_input_stream_test.cpp
+                    common/io/offset_output_stream_test.cpp
                     common/lookup/sort/sort_lookup_store_test.cpp
                     common/logging/logging_test.cpp
                     common/metrics/metrics_impl_test.cpp
@@ -829,6 +837,7 @@ if(PAIMON_BUILD_TESTS)
                     core/io/data_file_meta_test.cpp
                     core/io/data_file_writer_base_test.cpp
                     core/io/data_file_index_writer_test.cpp
+                    core/io/spillable_index_output_stream_test.cpp
                     core/io/file_index_options_test.cpp
                     core/io/file_index_evaluator_test.cpp
                     core/io/single_file_writer_test.cpp

src/paimon/common/defs.cpp

@@ -118,6 +118,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_FORMAT_VERSION[] = "file-index.format.version";
 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";

src/paimon/common/file_index/bitmap/bitmap_file_index.cpp

@@ -18,13 +18,15 @@
 
 #include "paimon/common/file_index/bitmap/bitmap_file_index.h"
 
+#include <limits>
 #include <utility>
 
 #include "arrow/c/bridge.h"
 #include "fmt/format.h"
 #include "paimon/common/file_index/bitmap/bitmap_file_index_meta.h"
 #include "paimon/common/file_index/bitmap/bitmap_file_index_meta_v1.h"
 #include "paimon/common/file_index/bitmap/bitmap_file_index_meta_v2.h"
+#include "paimon/common/io/offset_input_stream.h"
 #include "paimon/common/memory/memory_segment_utils.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
@@ -77,9 +79,15 @@ FieldType BitmapFileIndex::ConvertType(const FieldType& data_type) {
 }
 
 Result<std::shared_ptr<FileIndexReader>> BitmapFileIndex::CreateReader(
-    ::ArrowSchema* c_arrow_schema, int32_t start, int32_t length,
+    ::ArrowSchema* c_arrow_schema, int64_t start, int64_t length,
     const std::shared_ptr<InputStream>& input_stream,
     const std::shared_ptr<MemoryPool>& pool) const {
+    if (length > std::numeric_limits<int32_t>::max()) {
+        return Status::NotImplemented(
+            "Bitmap file index does not support payloads larger than INT32_MAX");
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OffsetInputStream> bounded_stream,
+                           OffsetInputStream::Create(input_stream, length, start));
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_schema,
                                       arrow::ImportSchema(c_arrow_schema));
     if (arrow_schema->num_fields() != 1) {
@@ -90,8 +98,7 @@ Result<std::shared_ptr<FileIndexReader>> BitmapFileIndex::CreateReader(
     auto arrow_type = arrow_schema->field(0)->type();
     PAIMON_ASSIGN_OR_RAISE(FieldType data_type,
                            FieldTypeUtils::ConvertToFieldType(arrow_type->id()));
-    return std::make_shared<BitmapFileIndexReader>(arrow_type, data_type, start, length,
-                                                   input_stream, pool);
+    return std::make_shared<BitmapFileIndexReader>(arrow_type, data_type, bounded_stream, pool);
 }
 
 Result<std::shared_ptr<FileIndexWriter>> BitmapFileIndex::CreateWriter(
@@ -228,16 +235,10 @@ Result<PAIMON_UNIQUE_PTR<Bytes>> BitmapFileIndexWriter::SerializedBytes() const
 }
 
 BitmapFileIndexReader::BitmapFileIndexReader(const std::shared_ptr<arrow::DataType>& arrow_type,
-                                             const FieldType& data_type, int32_t start,
-                                             int32_t length,
+                                             const FieldType& data_type,
                                              const std::shared_ptr<InputStream>& input_stream,
                                              const std::shared_ptr<MemoryPool>& pool)
-    : head_start_(start),
-      length_(length),
-      data_type_(data_type),
-      arrow_type_(arrow_type),
-      pool_(pool),
-      input_stream_(input_stream) {}
+    : data_type_(data_type), arrow_type_(arrow_type), pool_(pool), input_stream_(input_stream) {}
 
 Result<std::shared_ptr<FileIndexResult>> BitmapFileIndexReader::VisitEqual(const Literal& literal) {
     return VisitIn({literal});
@@ -329,16 +330,17 @@ Result<RoaringBitmap32> BitmapFileIndexReader::ReadBitmap(const Literal& literal
 
 Status BitmapFileIndexReader::ReadInternalMeta() {
     if (!bitmap_file_index_meta_) {
-        PAIMON_RETURN_NOT_OK(input_stream_->Seek(head_start_, SeekOrigin::FS_SEEK_SET));
+        PAIMON_RETURN_NOT_OK(input_stream_->Seek(0, SeekOrigin::FS_SEEK_SET));
+        PAIMON_ASSIGN_OR_RAISE(int64_t length, input_stream_->Length());
         DataInputStream data_input_stream(input_stream_);
         PAIMON_ASSIGN_OR_RAISE(int8_t version, data_input_stream.ReadValue<int8_t>());
         FieldType converted_type = BitmapFileIndex::ConvertType(data_type_);
         if (version == BitmapFileIndex::VERSION_1) {
             bitmap_file_index_meta_ = std::make_shared<BitmapFileIndexMetaV1>(
-                converted_type, head_start_, length_, pool_);
+                converted_type, /*start=*/0, static_cast<int32_t>(length), pool_);
         } else if (version == BitmapFileIndex::VERSION_2) {
-            bitmap_file_index_meta_ =
-                std::make_shared<BitmapFileIndexMetaV2>(converted_type, length_, pool_);
+            bitmap_file_index_meta_ = std::make_shared<BitmapFileIndexMetaV2>(
+                converted_type, static_cast<int32_t>(length), pool_);
         } else {
             return Status::Invalid(fmt::format("unknown bitmap file index version {}", version));
         }

src/paimon/common/file_index/bitmap/bitmap_file_index.h

@@ -46,7 +46,7 @@ class PAIMON_EXPORT BitmapFileIndex : public FileIndexer {
     ~BitmapFileIndex() override = default;
 
     Result<std::shared_ptr<FileIndexReader>> CreateReader(
-        ::ArrowSchema* arrow_schema, int32_t start, int32_t length,
+        ::ArrowSchema* arrow_schema, int64_t start, int64_t length,
         const std::shared_ptr<InputStream>& input_stream,
         const std::shared_ptr<MemoryPool>& pool) const override;
 
@@ -101,7 +101,7 @@ class BitmapFileIndexReader : public FileIndexReader,
                               public std::enable_shared_from_this<BitmapFileIndexReader> {
  public:
     BitmapFileIndexReader(const std::shared_ptr<arrow::DataType>& arrow_type,
-                          const FieldType& data_type, int32_t start, int32_t length,
+                          const FieldType& data_type,
                           const std::shared_ptr<InputStream>& input_stream,
                           const std::shared_ptr<MemoryPool>& pool);
 
@@ -123,8 +123,6 @@ class BitmapFileIndexReader : public FileIndexReader,
     Status ReadInternalMeta();
 
  private:
-    int32_t head_start_;
-    int32_t length_;
     FieldType data_type_;
     std::shared_ptr<arrow::DataType> arrow_type_;
     std::shared_ptr<MemoryPool> pool_;

src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp

@@ -27,6 +27,7 @@
 
 #include "arrow/c/bridge.h"
 #include "fmt/format.h"
+#include "paimon/common/io/offset_input_stream.h"
 #include "paimon/common/predicate/literal_converter.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/checked_cast.h"
@@ -43,9 +44,11 @@ class MemoryPool;
 BloomFilterFileIndex::BloomFilterFileIndex(const std::map<std::string, std::string>& options)
     : options_(options) {}
 Result<std::shared_ptr<FileIndexReader>> BloomFilterFileIndex::CreateReader(
-    ::ArrowSchema* c_arrow_schema, int32_t start, int32_t length,
+    ::ArrowSchema* c_arrow_schema, int64_t start, int64_t length,
     const std::shared_ptr<InputStream>& input_stream,
     const std::shared_ptr<MemoryPool>& pool) const {
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OffsetInputStream> bounded_stream,
+                           OffsetInputStream::Create(input_stream, length, start));
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_schema,
                                       arrow::ImportSchema(c_arrow_schema));
     if (arrow_schema->num_fields() != 1) {
@@ -55,10 +58,9 @@ Result<std::shared_ptr<FileIndexReader>> BloomFilterFileIndex::CreateReader(
     }
     auto arrow_type = arrow_schema->field(0)->type();
 
-    PAIMON_RETURN_NOT_OK(input_stream->Seek(start, SeekOrigin::FS_SEEK_SET));
     auto bytes = std::make_shared<Bytes>(length, pool.get());
     PAIMON_ASSIGN_OR_RAISE(int64_t actual_read_len,
-                           input_stream->Read(bytes->data(), bytes->size()));
+                           bounded_stream->Read(bytes->data(), bytes->size()));
     if (static_cast<size_t>(actual_read_len) != bytes->size()) {
         return Status::Invalid(
             fmt::format("create reader for BloomFilterFileIndex failed, expected read len "

src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.h

@@ -50,7 +50,7 @@ class BloomFilterFileIndex : public FileIndexer {
     ~BloomFilterFileIndex() override = default;
 
     Result<std::shared_ptr<FileIndexReader>> CreateReader(
-        ::ArrowSchema* arrow_schema, int32_t start, int32_t length,
+        ::ArrowSchema* arrow_schema, int64_t start, int64_t length,
         const std::shared_ptr<InputStream>& input_stream,
         const std::shared_ptr<MemoryPool>& pool) const override;