feat(realtime): improve deduplication, spill and index support

docs/source/user_guide.rst

@@ -36,6 +36,7 @@ User Guide
    user_guide/format_table
    user_guide/system_tables
    user_guide/write
+   user_guide/realtime
    user_guide/commit
    user_guide/compaction
    user_guide/read

include/paimon/file_index/file_indexer.h

@@ -23,11 +23,13 @@
 #include <string>
 #include <vector>
 
+#include "paimon/defs.h"
 #include "paimon/file_index/file_index_reader.h"
 #include "paimon/file_index/file_index_result.h"
 #include "paimon/file_index/file_index_writer.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/memory/memory_pool.h"
+#include "paimon/predicate/function.h"
 #include "paimon/visibility.h"
 
 struct ArrowSchema;
@@ -38,6 +40,13 @@ class PAIMON_EXPORT FileIndexer {
  public:
     virtual ~FileIndexer() = default;
 
+    /// Whether this indexer returns exact file-local matching row positions for the field type.
+    /// Supported evaluations return BitmapIndexResult (or Skip), never Remain or false positives.
+    /// The schema and values must satisfy the index's writer constraints.
+    virtual bool SupportsExactPredicate(Function::Type, FieldType) const {
+        return false;
+    }
+
     /// Extra columns required while building this index, in addition to the indexed column.
     virtual Result<std::optional<std::vector<std::string>>> GetExtraFieldNames() const {
         return std::optional<std::vector<std::string>>(std::nullopt);

include/paimon/realtime/realtime_store.h

@@ -95,6 +95,7 @@ struct PAIMON_EXPORT RealtimeWriteBatch {
 /// Row counts are physical stored rows. For primary-key stores they include old versions and
 /// delete records, rather than the rows visible after merge-on-read. A segment removed by
 /// `AdvanceCommittedOffset` is no longer included, even if an older read view still pins it.
+/// Deduplicate stores also include retained in-memory key-index buffers and offset bitmaps.
 struct PAIMON_EXPORT RealtimeStoreDataUsage {
     uint64_t building_memory_bytes = 0;
     uint64_t sealed_memory_bytes = 0;
@@ -211,6 +212,28 @@ class PAIMON_EXPORT RealtimeStore {
     virtual uint64_t GetMemoryUsage() const = 0;
 };
 
+/// Optional capability for `RealtimeStore` implementations to separate segment freezing from spill
+/// I/O.
+class PAIMON_EXPORT RealtimeStoreTwoPhaseSeal {
+ public:
+    virtual ~RealtimeStoreTwoPhaseSeal() = default;
+
+    /// Freezes the building segment without spill I/O.
+    virtual Result<std::optional<std::shared_ptr<RealtimeSegmentHandle>>>
+    FreezeBuildingSegment() = 0;
+
+    /// Spills a segment returned by FreezeBuildingSegment and returns its commit handle.
+    virtual Result<std::shared_ptr<RealtimeSegmentHandle>> SpillFrozenSegment(
+        const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0;
+
+    /// Advances the committed offset without destroying retired storage. The returned view owns
+    /// exactly the segments removed in this operation, including concurrent spill replacements.
+    /// The caller releases it after unlocking its write mutex, so cleanup never runs under that
+    /// mutex. This operation must not perform I/O.
+    virtual Result<std::shared_ptr<RealtimeReadView>> RetireCommittedSegments(
+        int64_t committed_end_offset) = 0;
+};
+
 /// Factory for application-provided `RealtimeStore` implementations.
 class PAIMON_EXPORT RealtimeStoreFactory {
  public:

src/paimon/CMakeLists.txt

@@ -423,6 +423,7 @@ set(PAIMON_CORE_SRCS
     core/realtime/data_file_key_offset_index_writer.cpp
     core/realtime/key_offset_index_utils.cpp
     core/realtime/realtime_key_offset_index.cpp
+    core/realtime/realtime_file_schema.cpp
     core/realtime/realtime_offset_file_index_lookup.cpp
     core/realtime/realtime_deduplicate_state.cpp
     core/realtime/realtime_deduplicate_writer.cpp
@@ -907,6 +908,7 @@ if(PAIMON_BUILD_TESTS)
                     core/realtime/key_offset_index_utils_test.cpp
                     core/realtime/primary_key_realtime_store_test.cpp
                     core/realtime/realtime_deduplicate_state_test.cpp
+                    core/realtime/realtime_file_schema_test.cpp
                     core/realtime/realtime_key_offset_index_test.cpp
                     core/realtime/realtime_offset_batch_reader_test.cpp
                     core/realtime/realtime_offset_file_index_lookup_test.cpp

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

@@ -46,6 +46,37 @@ class MemoryPool;
 BitmapFileIndex::BitmapFileIndex(const std::map<std::string, std::string>& options)
     : options_(options) {}
 
+bool BitmapFileIndex::SupportsExactPredicate(Function::Type function, FieldType field_type) const {
+    switch (field_type) {
+        case FieldType::BOOLEAN:
+        case FieldType::TINYINT:
+        case FieldType::SMALLINT:
+        case FieldType::INT:
+        case FieldType::BIGINT:
+        case FieldType::FLOAT:
+        case FieldType::DOUBLE:
+        case FieldType::STRING:
+        case FieldType::DATE:
+            break;
+        case FieldType::TIMESTAMP:
+            // Nanosecond timestamps are truncated to microseconds.
+            return function == Function::Type::IS_NULL || function == Function::Type::IS_NOT_NULL;
+        default:
+            return false;
+    }
+    switch (function) {
+        case Function::Type::EQUAL:
+        case Function::Type::NOT_EQUAL:
+        case Function::Type::IN:
+        case Function::Type::NOT_IN:
+        case Function::Type::IS_NULL:
+        case Function::Type::IS_NOT_NULL:
+            return true;
+        default:
+            return false;
+    }
+}
+
 Result<Literal> BitmapFileIndex::ConvertLiteral(
     const Literal& literal, const std::shared_ptr<arrow::DataType>& arrow_type) {
     if (literal.GetType() != FieldType::TIMESTAMP) {

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

@@ -45,6 +45,8 @@ class PAIMON_EXPORT BitmapFileIndex : public FileIndexer {
     explicit BitmapFileIndex(const std::map<std::string, std::string>& options);
     ~BitmapFileIndex() override = default;
 
+    bool SupportsExactPredicate(Function::Type function, FieldType field_type) const override;
+
     Result<std::shared_ptr<FileIndexReader>> CreateReader(
         ::ArrowSchema* arrow_schema, int64_t start, int64_t length,
         const std::shared_ptr<InputStream>& input_stream,

src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.cpp

@@ -65,6 +65,39 @@ class MemoryPool;
 BitSliceIndexBitmapFileIndex::BitSliceIndexBitmapFileIndex(
     const std::map<std::string, std::string>& options) {}
 
+bool BitSliceIndexBitmapFileIndex::SupportsExactPredicate(Function::Type function,
+                                                          FieldType field_type) const {
+    switch (field_type) {
+        case FieldType::TINYINT:
+        case FieldType::SMALLINT:
+        case FieldType::INT:
+        case FieldType::BIGINT:
+        case FieldType::DATE:
+        case FieldType::DECIMAL:
+            break;
+        case FieldType::TIMESTAMP:
+            // Nanosecond timestamps are truncated to microseconds.
+            return function == Function::Type::IS_NULL || function == Function::Type::IS_NOT_NULL;
+        default:
+            return false;
+    }
+    switch (function) {
+        case Function::Type::EQUAL:
+        case Function::Type::NOT_EQUAL:
+        case Function::Type::IN:
+        case Function::Type::NOT_IN:
+        case Function::Type::IS_NULL:
+        case Function::Type::IS_NOT_NULL:
+        case Function::Type::GREATER_THAN:
+        case Function::Type::GREATER_OR_EQUAL:
+        case Function::Type::LESS_THAN:
+        case Function::Type::LESS_OR_EQUAL:
+            return true;
+        default:
+            return false;
+    }
+}
+
 Result<std::shared_ptr<FileIndexWriter>> BitSliceIndexBitmapFileIndex::CreateWriter(
     ::ArrowSchema* c_arrow_schema, const std::shared_ptr<MemoryPool>& pool) const {
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> arrow_schema,

src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h

@@ -48,6 +48,8 @@ class BitSliceIndexBitmapFileIndex : public FileIndexer {
     explicit BitSliceIndexBitmapFileIndex(const std::map<std::string, std::string>& options);
     ~BitSliceIndexBitmapFileIndex() override = default;
 
+    bool SupportsExactPredicate(Function::Type function, FieldType field_type) const override;
+
     Result<std::shared_ptr<FileIndexReader>> CreateReader(
         ::ArrowSchema* arrow_schema, int64_t start, int64_t length,
         const std::shared_ptr<InputStream>& input_stream,

src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.cpp

@@ -40,6 +40,41 @@ namespace paimon {
 RangeBitmapFileIndex::RangeBitmapFileIndex(const std::map<std::string, std::string>& options)
     : options_(options) {}
 
+bool RangeBitmapFileIndex::SupportsExactPredicate(Function::Type function,
+                                                  FieldType field_type) const {
+    switch (field_type) {
+        case FieldType::BOOLEAN:
+        case FieldType::TINYINT:
+        case FieldType::SMALLINT:
+        case FieldType::INT:
+        case FieldType::BIGINT:
+        case FieldType::FLOAT:
+        case FieldType::DOUBLE:
+        case FieldType::STRING:
+        case FieldType::DATE:
+        case FieldType::TIMESTAMP:
+        case FieldType::DECIMAL:
+            break;
+        default:
+            return false;
+    }
+    switch (function) {
+        case Function::Type::EQUAL:
+        case Function::Type::NOT_EQUAL:
+        case Function::Type::IN:
+        case Function::Type::NOT_IN:
+        case Function::Type::IS_NULL:
+        case Function::Type::IS_NOT_NULL:
+        case Function::Type::GREATER_THAN:
+        case Function::Type::GREATER_OR_EQUAL:
+        case Function::Type::LESS_THAN:
+        case Function::Type::LESS_OR_EQUAL:
+            return true;
+        default:
+            return false;
+    }
+}
+
 Result<std::shared_ptr<FileIndexReader>> RangeBitmapFileIndex::CreateReader(
     ::ArrowSchema* arrow_schema, const int64_t start, const int64_t length,
     const std::shared_ptr<InputStream>& input_stream,

src/paimon/common/file_index/rangebitmap/range_bitmap_file_index.h

@@ -44,6 +44,8 @@ class PAIMON_EXPORT RangeBitmapFileIndex final : public FileIndexer {
 
     ~RangeBitmapFileIndex() override = default;
 
+    bool SupportsExactPredicate(Function::Type function, FieldType field_type) const override;
+
     Result<std::shared_ptr<FileIndexReader>> CreateReader(
         ::ArrowSchema* arrow_schema, int64_t start, int64_t length,
         const std::shared_ptr<InputStream>& input_stream,

src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp

@@ -19,10 +19,7 @@
 #include "paimon/common/reader/prefetch_file_batch_reader_impl.h"
 
 #include <atomic>
-#include <chrono>
-#include <condition_variable>
 #include <limits>
-#include <mutex>
 #include <set>
 
 #include "arrow/compute/api.h"
@@ -44,6 +41,7 @@
 #include "paimon/testing/mock/mock_file_batch_reader.h"
 #include "paimon/testing/mock/mock_file_system.h"
 #include "paimon/testing/mock/mock_format_reader_builder.h"
+#include "paimon/testing/utils/blocking_file_system.h"
 #include "paimon/testing/utils/read_result_collector.h"
 #include "paimon/testing/utils/testharness.h"
 
@@ -133,37 +131,6 @@ class FailingFileSystem : public MockFileSystem {
     }
 };
 
-/// Blocks every Open() until `opens_needed` of them are in flight at once, so a caller that
-/// opens its streams one at a time never reaches the barrier and fails instead of passing
-/// silently. The wait is bounded so that failure is reported rather than hanging the test.
-class BarrierFileSystem : public MockFileSystem {
- public:
-    explicit BarrierFileSystem(uint32_t opens_needed) : opens_needed_(opens_needed) {}
-
-    Result<std::unique_ptr<InputStream>> Open(const std::string& path) const override {
-        const uint32_t arrived = ++inflight_opens_;
-        std::unique_lock<std::mutex> lock(mutex_);
-        if (!condition_.wait_for(lock, kBarrierTimeout,
-                                 [this] { return inflight_opens_ >= opens_needed_; })) {
-            return Status::IOError(fmt::format(
-                "only {} of {} opens were in flight at once, so they are not concurrent", arrived,
-                opens_needed_));
-        }
-        condition_.notify_all();
-        return std::make_unique<MockInputStream>();
-    }
-
- private:
-    // Generous so a loaded machine does not turn this into a flaky test; a regression makes the
-    // opens serial, which no wait ever satisfies.
-    static constexpr std::chrono::seconds kBarrierTimeout{60};
-
-    const uint32_t opens_needed_;
-    mutable std::atomic<uint32_t> inflight_opens_{0};
-    mutable std::mutex mutex_;
-    mutable std::condition_variable condition_;
-};
-
 class IoReadingMockFileBatchReader : public MockFileBatchReader {
  public:
     IoReadingMockFileBatchReader(const std::shared_ptr<arrow::Array>& data,
@@ -1421,7 +1388,10 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestCreateOpensAllStreamsConcurrently) {
     const int32_t prefetch_max_parallel_num = 3;
     // One stream for the cache plus one per reader.
     const uint32_t opens_needed = prefetch_max_parallel_num + 1;
-    auto barrier_fs = std::make_shared<BarrierFileSystem>(opens_needed);
+    auto barrier_fs = std::make_shared<BlockingFileSystem>(std::make_shared<MockFileSystem>());
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<BlockingFileSystem::Gate> barrier,
+        barrier_fs->Barrier(BlockingFileSystem::Operation::OPEN, /*path=*/"", opens_needed));
     MockFormatReaderBuilder reader_builder(data_array, data_type_, batch_size);
     // Unlike executor_ above, this one has a thread per open: with fewer threads than opens the
     // barrier could not be reached however concurrent the dispatch is.
@@ -1436,6 +1406,7 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestCreateOpensAllStreamsConcurrently) {
             /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/true, CacheConfig(),
             /*enable_io_metrics=*/false, WarmupLevel::DECODED, pool_, GetArrowPool(pool_)));
     ASSERT_NE(reader, nullptr);
+    ASSERT_TRUE(barrier->WaitForArrival());
 }
 
 }  // namespace paimon::test