feat(realtime): support user-defined-key deduplication for append tables (#393)

Author: lxy-9602

include/paimon/defs.h

@@ -645,6 +645,10 @@ struct PAIMON_EXPORT Options {
     /// Supported values are "none" and "full". Default value is "none".
     static const char REALTIME_STORE_STATS_MODE[];
 
+    /// "realtime.deduplicate-key-fields" - Comma-separated user-defined deduplicate key fields
+    /// used for last-write-wins deduplication on an append table. No default value.
+    static const char REALTIME_DEDUPLICATE_KEY_FIELDS[];
+
     /// "scan.timestamp" - Optional timestamp string used in case of "from-timestamp" scan mode,
     /// as an alternative to "scan.timestamp-millis".
     /// It will be automatically converted to timestamp in unix milliseconds, using local time zone.

include/paimon/file_store_commit.h

@@ -98,6 +98,10 @@ class PAIMON_EXPORT FileStoreCommit {
     /// Concurrent rollback or partition deletion must be fenced by the upstream coordinator.
     /// Configure catalog writers with `WriteContextBuilder::WithCatalog()` too.
     ///
+    /// In deduplicate mode each commit attempt resolves logical deleted offsets against the
+    /// latest snapshot and merges them with its deletion vectors. Keep the complete progress
+    /// when retrying; serializing only CommitMessage discards logical deletions.
+    ///
     /// @param realtime_commits Commit messages and left-closed, right-open offset ranges to
     /// commit.
     /// @param commit_identifier Identifier of the streaming commit operation.

include/paimon/file_store_write.h

@@ -52,12 +52,16 @@ class PAIMON_EXPORT FileStoreWrite {
     virtual ~FileStoreWrite() = default;
 
     /// Support write an input `RecordBatch` to internal buffer or file.
-    /// @note Real-time writers require a non-nullable int64 `_REALTIME_OFFSET` field before the
-    ///       table write fields. Its values must be strictly increasing within each batch and
-    ///       monotonically increasing for each partition-bucket across batches; gaps are allowed.
-    ///       The field is used for snapshot progress and is not written to data files.
+    /// @note Real-time writers require a non-nullable int64 `_REALTIME_OFFSET`. Its values must be
+    ///       strictly increasing within each batch and monotonically increasing for each
+    ///       partition-bucket across batches; gaps are allowed. Append-only and primary-key modes
+    ///       receive it as an internal prefix and do not write it to data files. Deduplicate mode
+    ///       requires it in the table schema and persists it as a normal data-file column.
     /// @note If a field in table schema is marked as non-nullable (`nullable = false`),
     ///       the corresponding array in `batch` must have zero null entries.
+    /// @note A failure while writing or flushing a real-time batch invalidates the writer and its
+    ///       `RealtimeContext`. Recreate both and let upstream replay input from the durable
+    ///       recovery offset persisted in the snapshot; retrying the failed writer is unsupported.
     virtual Status Write(std::unique_ptr<RecordBatch>&& batch) = 0;
 
     /// Slices the current in-memory real-time data into a sealed segment so it can be reclaimed
@@ -111,7 +115,9 @@ class PAIMON_EXPORT FileStoreWrite {
     /// Each range is returned atomically with the commit message generated from the same sealed
     /// segments. Repeated calls return incremental progress. The upstream coordinator must retain
     /// every result until it is committed and include all earlier prepared-but-uncommitted
-    /// progress when a later checkpoint subsumes it.
+    /// progress when a later checkpoint subsumes it. In deduplicate mode the result also retains
+    /// an immutable deleted-offset bitmap; physical deletion vectors are generated at commit,
+    /// not at prepare. Preserve the complete progress, not just its CommitMessage.
     ///
     /// @param commit_identifier Identifier of this prepare-commit operation in streaming mode.
     /// @return Real-time commit messages with their partition-bucket offset ranges.

include/paimon/realtime/arrow_realtime_store_factory.h

@@ -26,7 +26,8 @@ namespace paimon {
 /// Factory for Paimon's default Arrow-backed `RealtimeStore`.
 class PAIMON_EXPORT ArrowRealtimeStoreFactory : public RealtimeStoreFactory {
  public:
-    /// Creates an Arrow-backed store for one partition and bucket.
+    /// Creates an Arrow-backed append or primary-key store for one partition and bucket.
+    /// The framework uses the append implementation as the delegate of its DEDUPLICATE wrapper.
     Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) override;
 };
 

include/paimon/realtime/realtime_commit_progress.h

@@ -24,6 +24,7 @@
 #include "paimon/commit_message.h"
 #include "paimon/realtime/offset_range.h"
 #include "paimon/realtime/realtime_context.h"
+#include "paimon/utils/roaring_bitmap64.h"
 #include "paimon/visibility.h"
 
 namespace paimon {
@@ -41,6 +42,10 @@ struct PAIMON_EXPORT RealtimeCommitProgress {
     RealtimePartitionBucket partition_bucket;
     /// Left-closed, right-open offset range represented by the commit message.
     OffsetRange offset_range;
+    /// Immutable logical deletions for deduplicate mode; null for other modes. Commit resolves
+    /// these offsets against the latest snapshot and rebuilds physical DVs on every retry.
+    /// This bitmap is not embedded in CommitMessage serialization.
+    std::shared_ptr<const RoaringBitmap64> deleted_offsets = nullptr;
 };
 
 }  // namespace paimon

include/paimon/realtime/realtime_store.h

@@ -46,20 +46,20 @@ class Predicate;
 enum class PAIMON_EXPORT RealtimeStoreMode {
     APPEND_ONLY,
     PRIMARY_KEY,
+    /// Experimental mode whose persisted data and index formats may change in future releases.
+    DEDUPLICATE,
 };
 
-/// Parameters used by a `RealtimeStoreFactory` to create a store.
+/// Parameters used to create a real-time store.
 struct PAIMON_EXPORT RealtimeStoreCreateRequest {
-    /// Schema whose ownership is transferred to the factory. Append mode receives the complete
-    /// append store-write schema: [_REALTIME_OFFSET, table write fields]. Primary-key mode receives
-    /// the real-time primary-key store-write schema:
-    /// [_SEQUENCE_NUMBER, _VALUE_KIND, _REALTIME_OFFSET, table write fields].
+    /// Physical write schema whose ownership is transferred to the consumer. Its layout is
+    /// mode-specific and contains a non-null int64 `_REALTIME_OFFSET` field.
     std::unique_ptr<::ArrowSchema> write_schema;
     /// Table options available to the store implementation.
     std::map<std::string, std::string> options;
     /// Memory pool for allocations retained by the store.
     std::shared_ptr<MemoryPool> memory_pool;
-    /// Table mode implemented by the store.
+    /// Requested store mode.
     RealtimeStoreMode mode = RealtimeStoreMode::APPEND_ONLY;
     /// Statistics collected by the store for query pruning.
     StatisticsMode statistics_mode = StatisticsMode::NONE;
@@ -69,6 +69,9 @@ struct PAIMON_EXPORT RealtimeStoreCreateRequest {
     /// File system used to access `temp_directory`. The default store retains shared ownership
     /// for its lifetime. Custom stores may ignore this hint.
     std::shared_ptr<FileSystem> file_system = nullptr;
+    /// User-defined deduplicate key field names when key-based deduplication is configured; empty
+    /// otherwise.
+    std::vector<std::string> deduplicate_key_fields = {};
 };
 
 /// A record batch and its application-assigned offset bounds.
@@ -77,9 +80,9 @@ struct PAIMON_EXPORT RealtimeStoreCreateRequest {
 /// [_REALTIME_OFFSET, table write fields], and offsets are strictly increasing before the batch
 /// enters the store. Primary-key batches use the real-time primary-key store-write schema, are
 /// sorted by full primary key then sequence number, and retain the original offset in
-/// `_REALTIME_OFFSET`. `offset_range` is the left-closed, right-open envelope from the first
-/// application offset through one past the last; offsets may have gaps, so its count is not the
-/// batch row count.
+/// `_REALTIME_OFFSET`.
+/// `offset_range` is the left-closed, right-open envelope from the first application offset through
+/// one past the last; offsets may have gaps, so its count is not the batch row count.
 struct PAIMON_EXPORT RealtimeWriteBatch {
     /// Input batch whose ownership is transferred to `RealtimeStore::Write`.
     std::unique_ptr<RecordBatch> batch;
@@ -170,9 +173,8 @@ class PAIMON_EXPORT RealtimeStore {
     /// Creates readers that expose all rows in a sealed segment for Paimon file writing.
     ///
     /// The returned readers collectively expose every sealed row exactly once. Append-mode readers
-    /// preserve write order and contain `_REALTIME_OFFSET` followed by the table write fields.
-    /// Primary-key readers contain the real-time primary-key store fields; each reader's complete
-    /// stream is sorted by full primary key then sequence number.
+    /// preserve write order. Primary-key reader streams are sorted by full primary key then
+    /// sequence number. Returned batches match the mode-specific physical write schema.
     /// Returned readers have independent mutable read state and may be operated concurrently with
     /// one another without external synchronization.
     virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders(
@@ -215,7 +217,6 @@ class PAIMON_EXPORT RealtimeStoreFactory {
     virtual ~RealtimeStoreFactory() = default;
 
     /// Creates a store configured with the supplied schema, statistics, options, and memory pool.
-    /// Creates a store for the requested table mode.
     /// The factory consumes `request`, including ownership of `request.write_schema`.
     virtual Result<std::shared_ptr<RealtimeStore>> Create(RealtimeStoreCreateRequest&& request) = 0;
 };

src/paimon/CMakeLists.txt

@@ -86,6 +86,7 @@ set(PAIMON_COMMON_SRCS
     common/global_index/bitmap_global_index_result.cpp
     common/global_index/global_index_result.cpp
     common/global_index/global_indexer_factory.cpp
+    common/global_index/key_serializer.cpp
     common/io/buffered_input_stream.cpp
     common/io/byte_array_input_stream.cpp
     common/io/byte_array_output_stream.cpp
@@ -402,7 +403,12 @@ set(PAIMON_CORE_SRCS
     core/operation/write_context.cpp
     core/operation/write_restore.cpp
     core/realtime/arrow_realtime_store.cpp
+    core/realtime/arrow_deduplicate_realtime_store.cpp
     core/realtime/arrow_realtime_store_factory.cpp
+    core/realtime/realtime_key_offset_index.cpp
+    core/realtime/realtime_offset_file_index_lookup.cpp
+    core/realtime/realtime_deduplicate_state.cpp
+    core/realtime/realtime_deduplicate_writer.cpp
     core/realtime/realtime_offset_batch_reader.cpp
     core/realtime/realtime_primary_key_reader.cpp
     core/realtime/realtime_append_only_writer.cpp
@@ -850,14 +856,19 @@ if(PAIMON_BUILD_TESTS)
                     core/manifest/index_manifest_entry_serializer_test.cpp
                     core/manifest/index_manifest_file_handler_test.cpp
                     core/memory/writer_memory_manager_test.cpp
+                    core/realtime/arrow_deduplicate_realtime_store_test.cpp
                     core/realtime/arrow_realtime_store_test.cpp
                     core/realtime/primary_key_realtime_store_test.cpp
+                    core/realtime/realtime_deduplicate_state_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
                     core/realtime/realtime_store_read_pipeline_test.cpp
                     core/realtime/realtime_schema_layout_test.cpp
                     core/realtime/realtime_primary_key_reader_test.cpp
                     core/realtime/realtime_context_test.cpp
                     core/realtime/realtime_reader_test.cpp
+                    core/realtime/realtime_utils_test.cpp
                     core/mergetree/levels_test.cpp
                     core/mergetree/lookup_file_test.cpp
                     core/mergetree/lookup_levels_test.cpp

src/paimon/common/defs.cpp

@@ -170,6 +170,7 @@ const char Options::REALTIME_ENABLED[] = "realtime.enabled";
 const char Options::REALTIME_SPILL_ENABLED[] = "realtime.spill-enabled";
 const char Options::REALTIME_READ_VIEW_TTL[] = "realtime.read-view-ttl";
 const char Options::REALTIME_STORE_STATS_MODE[] = "realtime.store.stats-mode";
+const char Options::REALTIME_DEDUPLICATE_KEY_FIELDS[] = "realtime.deduplicate-key-fields";
 const char Options::SCAN_TIMESTAMP[] = "scan.timestamp";
 const char Options::SCAN_TAG_NAME[] = "scan.tag-name";
 const char Options::WRITE_ONLY[] = "write-only";

src/paimon/common/global_index/CMakeLists.txt

@@ -30,7 +30,6 @@ set(PAIMON_GLOBAL_INDEX_SRC
     sorted_file_meta_selector.cpp
     sorted_index_file_meta.cpp
     btree/lazy_filtered_btree_reader.cpp
-    key_serializer.cpp
     rangebitmap/range_bitmap_global_index.cpp
     rangebitmap/range_bitmap_global_index_factory.cpp
     offset_global_index_reader.cpp

src/paimon/common/global_index/key_serializer.h

@@ -26,10 +26,11 @@
 #include "paimon/memory/bytes.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/predicate/literal.h"
+#include "paimon/visibility.h"
 namespace paimon {
 
 /// Provides core methods to serialize, deserialize, and compare global index keys.
-class KeySerializer {
+class PAIMON_EXPORT KeySerializer {
  public:
     ~KeySerializer() = default;