feat(realtime): unify Arrow storage and support PK deletion vectors

include/paimon/file_store_write.h

@@ -60,6 +60,15 @@ class PAIMON_EXPORT FileStoreWrite {
     ///       the corresponding array in `batch` must have zero null entries.
     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
+    /// independently. A future implementation will support spilling sealed segments to a
+    /// temporary directory; currently this method only creates the in-memory segment boundary.
+    /// Calling this method on a non-real-time writer returns an error.
+    /// If sealing fails, the caller must recreate both the `RealtimeContext` and writer. The
+    /// upstream must then recover input from the durable recovery offset persisted in the
+    /// snapshot. Reusing the failed writer is unsupported.
+    virtual Status Seal();
+
     /// Compact data stored in given partition and bucket. Note that compaction process is only
     /// submitted and may not be completed when the method returns.
     ///
@@ -99,13 +108,17 @@ class PAIMON_EXPORT FileStoreWrite {
     /// Generates commit messages together with partition-bucket real-time offset ranges.
     ///
     /// Each range is returned atomically with the commit message generated from the same sealed
-    /// segment. Repeated calls return incremental progress. The upstream coordinator must retain
+    /// 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.
     ///
     /// @param commit_identifier Identifier of this prepare-commit operation in streaming mode.
     /// @return Real-time commit messages with their partition-bucket offset ranges.
     /// @note Calling this method on a non-real-time writer or in batch mode returns an error.
+    /// @note If preparation fails, the caller must recreate both the `RealtimeContext` and writer.
+    ///       The upstream must then recover input from the durable recovery offset persisted in the
+    ///       snapshot. The failed writer may contain partially prepared bucket state and must not
+    ///       be reused.
     virtual Result<std::vector<RealtimeCommitProgress>> PrepareCommitWithProgress(
         int64_t commit_identifier);
 
@@ -122,6 +135,12 @@ class PAIMON_EXPORT FileStoreWrite {
     virtual Status RefreshCommittedSnapshot(int64_t snapshot_id);
 
     virtual std::shared_ptr<Metrics> GetMetrics() const = 0;
+
+    /// Releases resources owned by this writer.
+    ///
+    /// Closing a real-time writer with data not covered by a successful
+    /// `PrepareCommitWithProgress()` returns an error and invalidates its `RealtimeContext`.
+    /// The caller must rebuild both objects and recover input from the durable snapshot offset.
     virtual Status Close() = 0;
 };
 

include/paimon/realtime/realtime_commit_progress.h

@@ -35,9 +35,9 @@ namespace paimon {
 /// count is not the represented row count. The progress fields are not embedded in `CommitMessage`
 /// serialization.
 struct PAIMON_EXPORT RealtimeCommitProgress {
-    /// Paimon commit message generated from one sealed segment.
+    /// Paimon commit message generated from one or more sealed segments.
     std::shared_ptr<CommitMessage> commit_message;
-    /// Partition-bucket containing the sealed segment.
+    /// Partition-bucket containing the sealed segments.
     RealtimePartitionBucket partition_bucket;
     /// Left-closed, right-open offset range represented by the commit message.
     OffsetRange offset_range;

include/paimon/realtime/realtime_store.h

@@ -50,9 +50,9 @@ enum class PAIMON_EXPORT RealtimeStoreMode {
 /// Parameters used by a `RealtimeStoreFactory` to create a store.
 struct PAIMON_EXPORT RealtimeStoreCreateRequest {
     /// Schema whose ownership is transferred to the factory. Append mode receives the complete
-    /// append transport schema: [_REALTIME_OFFSET, table write fields]. Primary-key mode receives
-    /// the realtime primary-key transport schema:
-    /// [_VALUE_KIND, _SEQUENCE_NUMBER, _REALTIME_OFFSET, table write fields].
+    /// 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].
     std::unique_ptr<::ArrowSchema> write_schema;
     /// Table options available to the store implementation.
     std::map<std::string, std::string> options;
@@ -66,12 +66,13 @@ struct PAIMON_EXPORT RealtimeStoreCreateRequest {
 
 /// A record batch and its application-assigned offset bounds.
 ///
-/// Append-mode batches use the append transport schema [_REALTIME_OFFSET, table write fields], and
-/// offsets are strictly increasing before the batch enters the store. Primary-key batches use the
-/// realtime primary-key transport 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.
+/// Append-mode batches use the append store-write schema
+/// [_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.
 struct PAIMON_EXPORT RealtimeWriteBatch {
     /// Input batch whose ownership is transferred to `RealtimeStore::Write`.
     std::unique_ptr<RecordBatch> batch;
@@ -109,11 +110,11 @@ class PAIMON_EXPORT RealtimeReadView {
 
 /// Parameters used by a `RealtimeStore` to create readers for a query.
 struct PAIMON_EXPORT RealtimeQueryContext {
-    /// Physical source schema the store must materialize. Query readers must include the mandatory
-    /// `_VALUE_KIND` field in returned batches. Paimon may subsequently convert physical fields
-    /// into the query's logical output schema, for example for selected-key MAP or VARIANT access.
-    /// This schema is borrowed and remains valid only during `CreateQueryReaders`; plugins must
-    /// import or copy it synchronously.
+    /// Physical source schema the store must materialize. Every returned batch must match this
+    /// schema exactly. Paimon may subsequently add framework fields or convert physical fields into
+    /// the query's logical output schema, for example for selected-key MAP or VARIANT access. This
+    /// schema is borrowed and remains valid only during `CreateQueryReaders`; plugins must import
+    /// or copy it synchronously.
     ::ArrowSchema* read_schema;
     /// Optional predicate using field indexes from `read_schema`. A non-null predicate allows the
     /// plugin to prune candidate rows. Exact filtering is applied by the Paimon read framework.
@@ -146,9 +147,9 @@ 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 `_VALUE_KIND`, `_REALTIME_OFFSET`, and table write fields.
-    /// Primary-key readers contain the realtime primary-key transport fields; each reader's
-    /// complete stream is sorted by full primary key then sequence number.
+    /// 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.
     virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders(
         const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0;
 

src/paimon/CMakeLists.txt

@@ -389,10 +389,10 @@ set(PAIMON_CORE_SRCS
     core/realtime/arrow_realtime_store_factory.cpp
     core/realtime/realtime_offset_batch_reader.cpp
     core/realtime/realtime_primary_key_reader.cpp
-    core/realtime/primary_key_realtime_store.cpp
     core/realtime/realtime_append_only_writer.cpp
     core/realtime/realtime_context.cpp
     core/realtime/realtime_context_impl.cpp
+    core/realtime/realtime_schema_layout.cpp
     core/realtime/realtime_store_read_pipeline.cpp
     core/realtime/realtime_primary_key_writer.cpp
     core/postpone/postpone_bucket_writer.cpp
@@ -804,6 +804,7 @@ if(PAIMON_BUILD_TESTS)
                     core/realtime/primary_key_realtime_store_test.cpp
                     core/realtime/realtime_offset_batch_reader_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

src/paimon/common/reader/complete_row_kind_batch_reader.cpp

@@ -72,13 +72,17 @@ Result<BatchReader::ReadBatchWithBitmap> CompleteRowKindBatchReader::NextBatchWi
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> row_kind_array,
                            PrepareRowKindArray(struct_array->length()));
     // complete row kind
-    UpdateFieldNamesWithRowKind(struct_array);
     arrow::ArrayVector fields_with_row_kind = {row_kind_array};
     fields_with_row_kind.insert(fields_with_row_kind.end(), struct_array->fields().begin(),
                                 struct_array->fields().end());
+    arrow::FieldVector schema_fields_with_row_kind = {
+        arrow::field(SpecialFields::ValueKind().Name(), arrow::int8())};
+    const arrow::FieldVector& schema_fields = struct_array->struct_type()->fields();
+    schema_fields_with_row_kind.insert(schema_fields_with_row_kind.end(), schema_fields.begin(),
+                                       schema_fields.end());
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
         std::shared_ptr<arrow::StructArray> array_with_row_kind,
-        arrow::StructArray::Make(fields_with_row_kind, field_names_with_row_kind_));
+        arrow::StructArray::Make(fields_with_row_kind, schema_fields_with_row_kind));
     PAIMON_RETURN_NOT_OK_FROM_ARROW(
         arrow::ExportArray(*array_with_row_kind, c_array.get(), c_schema.get()));
     PAIMON_RETURN_NOT_OK(AddArrowArrayLifetime(c_array.get(), c_schema.get(), arrow_pool_));
@@ -99,19 +103,4 @@ Result<std::shared_ptr<arrow::Array>> CompleteRowKindBatchReader::PrepareRowKind
     }
 }
 
-void CompleteRowKindBatchReader::UpdateFieldNamesWithRowKind(
-    const std::shared_ptr<arrow::StructArray>& struct_array) {
-    if (static_cast<size_t>(struct_array->struct_type()->num_fields()) + 1 ==
-        field_names_with_row_kind_.size()) {
-        return;
-    }
-    field_names_with_row_kind_.clear();
-    const auto& fields = struct_array->struct_type()->fields();
-    field_names_with_row_kind_.reserve(fields.size() + 1);
-    field_names_with_row_kind_.push_back(SpecialFields::ValueKind().Name());
-    for (const auto& field : fields) {
-        field_names_with_row_kind_.push_back(field->name());
-    }
-}
-
 }  // namespace paimon

src/paimon/common/reader/complete_row_kind_batch_reader.h

@@ -21,9 +21,7 @@
 
 #include <cstdint>
 #include <memory>
-#include <string>
 #include <utility>
-#include <vector>
 
 #include "arrow/api.h"
 #include "arrow/array/array_base.h"
@@ -52,7 +50,6 @@ class CompleteRowKindBatchReader : public BatchReader {
     void Close() override {
         reader_->Close();
         row_kind_array_.reset();
-        field_names_with_row_kind_.clear();
     }
 
     std::shared_ptr<Metrics> GetReaderMetrics() const override {
@@ -62,12 +59,9 @@ class CompleteRowKindBatchReader : public BatchReader {
  private:
     Result<std::shared_ptr<arrow::Array>> PrepareRowKindArray(int32_t struct_array_length);
 
-    void UpdateFieldNamesWithRowKind(const std::shared_ptr<arrow::StructArray>& struct_array);
-
  private:
     std::shared_ptr<arrow::MemoryPool> arrow_pool_;
     std::unique_ptr<BatchReader> reader_;
     std::shared_ptr<arrow::Array> row_kind_array_;
-    std::vector<std::string> field_names_with_row_kind_;
 };
 }  // namespace paimon

src/paimon/core/append/append_compact_coordinator.cpp

@@ -165,7 +165,8 @@ std::unique_ptr<AppendOnlyFileStoreWrite> CreateFileStoreWrite(
         path_factory, snapshot_manager, schema_manager,
         /*commit_user=*/"compact-coordinator",
         /*root_path=*/table_path, table_schema, arrow_schema,
-        /*write_schema=*/arrow_schema, partition_schema,
+        /*write_schema=*/arrow_schema,
+        /*realtime_schema_layout=*/nullptr, partition_schema,
         /*dv_maintainer_factory=*/nullptr,
         /*io_manager=*/nullptr, core_options,
         /*ignore_previous_files=*/true,