refactor a little

Author: lxy-9602

include/paimon/file_store_write.h

@@ -60,7 +60,9 @@ class PAIMON_EXPORT FileStoreWrite {
     ///       the corresponding array in `batch` must have zero null entries.
     virtual Status Write(std::unique_ptr<RecordBatch>&& batch) = 0;
 
-    /// Seals the current in-memory real-time segment without flushing it to a data file.
+    /// 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.
     virtual Status Seal();
 

include/paimon/realtime/realtime_store.h

@@ -110,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.
@@ -147,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 real-time primary-key store 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/core/operation/key_value_file_store_write_test.cpp

@@ -468,12 +468,10 @@ TEST_F(KeyValueFileStoreWriteTest, TestRealtimeWrite) {
     ASSERT_NOK_WITH_MSG(writer->Write(MakeBatch(realtime_schema, R"([[30, 3, "backwards"]])")),
                         "offset moved backwards or was duplicated");
     ASSERT_NOK(writer->Write(MakeBatch(realtime_schema, R"([[null, 3, "null-offset"]])")));
-    using RealtimePrimaryKeyStoreRow = std::tuple<int8_t, int64_t, std::string, int64_t, int64_t>;
-    ASSERT_OK_AND_ASSIGN(std::vector<RealtimePrimaryKeyStoreRow> store_rows,
-                         ReadRealtimePrimaryKeyStoreRows(realtime_context));
-    ASSERT_EQ((std::vector<RealtimePrimaryKeyStoreRow>{
-                  {0, 1, "old", 0, 10}, {2, 1, "new", 2, 30}, {3, 2, "two", 1, 20}}),
-              store_rows);
+    ASSERT_OK_AND_ASSIGN(auto store_rows, ReadRealtimePrimaryKeyStoreRows(realtime_context));
+    ASSERT_EQ(
+        (decltype(store_rows){{0, 1, "old", 0, 10}, {2, 1, "new", 2, 30}, {3, 2, "two", 1, 20}}),
+        store_rows);
     ASSERT_OK_AND_ASSIGN(std::vector<RealtimeCommitProgress> progresses,
                          writer->PrepareCommitWithProgress(0));
     ASSERT_EQ(1, progresses.size());
@@ -523,10 +521,8 @@ TEST_F(KeyValueFileStoreWriteTest, TestRealtimePool) {
     ASSERT_GT(pool->allocation_count, allocations_before_write);
     ASSERT_OK(writer->Close());
     writer.reset();
-    using RealtimePrimaryKeyStoreRow = std::tuple<int8_t, int64_t, std::string, int64_t, int64_t>;
-    ASSERT_OK_AND_ASSIGN(std::vector<RealtimePrimaryKeyStoreRow> retained_rows,
-                         ReadRealtimePrimaryKeyStoreRows(realtime_context));
-    ASSERT_EQ((std::vector<RealtimePrimaryKeyStoreRow>{{0, 1, "one", 0, 0}}), retained_rows);
+    ASSERT_OK_AND_ASSIGN(auto retained_rows, ReadRealtimePrimaryKeyStoreRows(realtime_context));
+    ASSERT_EQ((decltype(retained_rows){{0, 1, "one", 0, 0}}), retained_rows);
 
     std::shared_ptr<TestingMemoryPool> rejecting_pool = std::make_shared<TestingMemoryPool>();
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> rejecting_context,
@@ -546,8 +542,7 @@ TEST_F(KeyValueFileStoreWriteTest, TestRealtimePool) {
     ASSERT_NOK_WITH_MSG(rejecting_writer->Write(MakeBatch(realtime_schema, R"([[0, 2, "two"]])")),
                         "Out of memory");
     ASSERT_GT(rejecting_pool->allocation_count, rejecting_allocations_before_write);
-    ASSERT_OK_AND_ASSIGN(std::vector<RealtimePrimaryKeyStoreRow> rejected_rows,
-                         ReadRealtimePrimaryKeyStoreRows(rejecting_context));
+    ASSERT_OK_AND_ASSIGN(auto rejected_rows, ReadRealtimePrimaryKeyStoreRows(rejecting_context));
     ASSERT_TRUE(rejected_rows.empty());
     ASSERT_OK(rejecting_writer->Close());
 }
@@ -606,18 +601,14 @@ TEST_F(KeyValueFileStoreWriteTest, TestRealtimeLimits) {
                          FileStoreWrite::Create(std::move(write_context)));
     ASSERT_OK(
         writer->Write(MakeBatch(realtime_schema, fmt::format(R"([[{}, 1, "legal"]])", max - 1))));
-    using RealtimePrimaryKeyStoreRow = std::tuple<int8_t, int64_t, std::string, int64_t, int64_t>;
-    ASSERT_OK_AND_ASSIGN(std::vector<RealtimePrimaryKeyStoreRow> store_rows,
-                         ReadRealtimePrimaryKeyStoreRows(realtime_context));
-    ASSERT_EQ((std::vector<RealtimePrimaryKeyStoreRow>{{0, 1, "legal", max - 1, max - 1}}),
-              store_rows);
+    ASSERT_OK_AND_ASSIGN(auto store_rows, ReadRealtimePrimaryKeyStoreRows(realtime_context));
+    ASSERT_EQ((decltype(store_rows){{0, 1, "legal", max - 1, max - 1}}), store_rows);
 
     ASSERT_NOK_WITH_MSG(
         writer->Write(MakeBatch(realtime_schema, fmt::format(R"([[{}, 2, "overflow"]])", max))),
         "real-time offset range exceeds INT64_MAX");
     ASSERT_OK_AND_ASSIGN(store_rows, ReadRealtimePrimaryKeyStoreRows(realtime_context));
-    ASSERT_EQ((std::vector<RealtimePrimaryKeyStoreRow>{{0, 1, "legal", max - 1, max - 1}}),
-              store_rows);
+    ASSERT_EQ((decltype(store_rows){{0, 1, "legal", max - 1, max - 1}}), store_rows);
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContextImpl> context_impl,
                          RealtimeContextImpl::Cast(realtime_context));
     ASSERT_OK_AND_ASSIGN(RealtimeReadState read_state, context_impl->AcquireReadState());

src/paimon/core/realtime/arrow_realtime_store.cpp

@@ -30,7 +30,6 @@
 #include "paimon/common/data/columnar/columnar_row.h"
 #include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/predicate/predicate_filter.h"
-#include "paimon/common/reader/complete_row_kind_batch_reader.h"
 #include "paimon/common/reader/reader_utils.h"
 #include "paimon/common/utils/arrow/arrow_utils.h"
 #include "paimon/common/utils/arrow/mem_utils.h"
@@ -424,10 +423,7 @@ Result<std::vector<std::unique_ptr<BatchReader>>> ArrowRealtimeStore::CreateComm
     }
     std::vector<std::unique_ptr<BatchReader>> readers;
     if (mode_ == RealtimeStoreMode::APPEND_ONLY) {
-        std::unique_ptr<BatchReader> reader =
-            std::make_unique<AppendCommitBatchReader>(arrow_segment);
-        readers.push_back(
-            std::make_unique<CompleteRowKindBatchReader>(std::move(reader), arrow_pool_));
+        readers.push_back(std::make_unique<AppendCommitBatchReader>(arrow_segment));
         return readers;
     }
     readers.reserve(arrow_segment->GetBatches().size());
@@ -486,7 +482,6 @@ Result<std::vector<std::unique_ptr<BatchReader>>> ArrowRealtimeStore::CreateQuer
     std::unique_ptr<BatchReader> reader = std::make_unique<AppendQueryBatchReader>(
         arrow_view, read_schema, predicate_filter, std::move(statistics_mapping), arrow_pool_,
         memory_pool_);
-    reader = std::make_unique<CompleteRowKindBatchReader>(std::move(reader), arrow_pool_);
     readers.push_back(std::move(reader));
     return readers;
 }

src/paimon/core/realtime/arrow_realtime_store_test.cpp

@@ -173,6 +173,19 @@ TEST_F(ArrowRealtimeStoreTest, TestQueryReaderClipsCommittedOffsetWithBitmap) {
         DataField::ConvertDataFieldToArrowField(SpecialFields::RealtimeOffset()),
         arrow::field("value", arrow::utf8()),
     });
+    {
+        std::unique_ptr<ArrowSchema> c_schema = MakeReadSchema(read_schema);
+        RealtimeQueryContext context{c_schema.get(), /*predicate=*/nullptr};
+        ASSERT_OK_AND_ASSIGN(std::vector<std::unique_ptr<BatchReader>> readers,
+                             store_->CreateQueryReaders(view, context));
+        ASSERT_OK_AND_ASSIGN(BatchReader::ReadBatchWithBitmap batch,
+                             readers[0]->NextBatchWithBitmap());
+        arrow::Result<std::shared_ptr<arrow::Array>> import_result =
+            arrow::ImportArray(batch.first.first.get(), batch.first.second.get());
+        ASSERT_TRUE(import_result.ok());
+        std::shared_ptr<arrow::Array> array = std::move(import_result).ValueOrDie();
+        ASSERT_TRUE(array->type()->Equals(arrow::struct_(read_schema->fields())));
+    }
     {
         std::unique_ptr<ArrowSchema> c_schema = MakeReadSchema(read_schema);
         RealtimeQueryContext context{c_schema.get(), /*predicate=*/nullptr};
@@ -236,15 +249,9 @@ TEST_F(ArrowRealtimeStoreTest, TestCommitReaderPreservesSlicedBatch) {
         arrow::ImportArray(batch.first.get(), batch.second.get());
     ASSERT_TRUE(import_result.ok()) << import_result.status().ToString();
     std::shared_ptr<arrow::Array> actual_array = std::move(import_result).ValueOrDie();
-    std::shared_ptr<arrow::DataType> expected_type = arrow::struct_({
-        arrow::field("_VALUE_KIND", arrow::int8()),
-        DataField::ConvertDataFieldToArrowField(SpecialFields::RealtimeOffset()),
-        arrow::field("id", arrow::int64()),
-        arrow::field("value", arrow::utf8()),
-    });
+    std::shared_ptr<arrow::DataType> expected_type = arrow::struct_(schema_->fields());
     std::shared_ptr<arrow::Array> expected_array =
-        arrow::ipc::internal::json::ArrayFromJSON(expected_type,
-                                                  R"([[0, 0, 1, null], [0, 1, 2, "c"]])")
+        arrow::ipc::internal::json::ArrayFromJSON(expected_type, R"([[0, 1, null], [1, 2, "c"]])")
             .ValueOrDie();
     ASSERT_TRUE(actual_array->Equals(*expected_array))
         << "expected: " << expected_array->ToString() << ", actual: " << actual_array->ToString();

src/paimon/core/realtime/primary_key_realtime_store_test.cpp

@@ -176,9 +176,7 @@ TEST(PrimaryKeyRealtimeStoreTest, TestCommitReaderPerStoredBatch) {
         std::shared_ptr<arrow::Array> expected,
         ArrayFromJson(actual->type(),
                       R"([[6, 1, 1, 1, "before"], [5, 0, 0, 3, "three"], [7, 2, 2, 2, "after"]])"));
-    ASSERT_TRUE(expected->Equals(actual)) << "expected:\n"
-                                          << expected->ToString() << "\nactual:\n"
-                                          << actual->ToString();
+    ASSERT_TRUE(expected->Equals(actual));
 }
 
 void AssertSlicedBatch(BatchReader* reader) {
@@ -283,9 +281,7 @@ TEST(PrimaryKeyRealtimeStoreTest, TestReclaimKeepsReadView) {
         std::shared_ptr<arrow::Array> expected,
         ArrayFromJson(actual->type(),
                       R"([[0, 0, 4, 1, "one"], [1, 0, 5, 2, "two"], [2, 0, 6, 3, "three"]])"));
-    ASSERT_TRUE(expected->Equals(actual)) << "expected:\n"
-                                          << expected->ToString() << "\nactual:\n"
-                                          << actual->ToString();
+    ASSERT_TRUE(expected->Equals(actual));
 }
 
 TEST(PrimaryKeyRealtimeStoreTest, TestQueryReaderPerStoredBatch) {
@@ -308,9 +304,7 @@ TEST(PrimaryKeyRealtimeStoreTest, TestQueryReaderPerStoredBatch) {
     ASSERT_OK_AND_ASSIGN(
         std::shared_ptr<arrow::Array> expected,
         ArrayFromJson(actual->type(), R"([[1, 0, 0, 2, "two"], [2, 0, 1, 1, "one"]])"));
-    ASSERT_TRUE(expected->Equals(actual)) << "expected:\n"
-                                          << expected->ToString() << "\nactual:\n"
-                                          << actual->ToString();
+    ASSERT_TRUE(expected->Equals(actual));
 }
 
 TEST(PrimaryKeyRealtimeStoreTest, TestQueryBatchOutlivesStoreAndReader) {

src/paimon/core/realtime/realtime_append_only_writer.cpp

@@ -170,23 +170,6 @@ Status RealtimeAppendOnlyWriter::FlushSegment(
         }
         std::shared_ptr<arrow::StructArray> struct_array =
             checked_pointer_cast<arrow::StructArray>(imported);
-        std::shared_ptr<arrow::Array> value_kind =
-            struct_array->GetFieldByName(SpecialFields::ValueKind().Name());
-        if (!value_kind || value_kind->type_id() != arrow::Type::INT8) {
-            return Status::Invalid(
-                "real-time store commit reader must return an INT8 _VALUE_KIND field");
-        }
-        std::shared_ptr<arrow::Int8Array> row_kinds =
-            checked_pointer_cast<arrow::Int8Array>(value_kind);
-        for (int64_t i = 0; i < row_kinds->length(); ++i) {
-            if (row_kinds->IsNull(i) ||
-                row_kinds->Value(i) != static_cast<int8_t>(RecordBatch::RowKind::INSERT)) {
-                return Status::Invalid(
-                    "append real-time store commit reader returned a non-INSERT row");
-            }
-        }
-        PAIMON_ASSIGN_OR_RAISE(struct_array, ArrowUtils::RemoveFieldFromStructArray(
-                                                 struct_array, SpecialFields::ValueKind().Name()));
         PAIMON_ASSIGN_OR_RAISE(struct_array,
                                ArrowUtils::RemoveFieldFromStructArray(
                                    struct_array, SpecialFields::RealtimeOffset().Name()));

src/paimon/core/realtime/realtime_schema_layout.cpp

@@ -62,7 +62,7 @@ RealtimeSchemaLayout::RealtimeSchemaLayout(RealtimeStoreMode mode,
     : user_schema_(user_schema), input_schema_(Prepend({RealtimeOffsetField()}, user_schema_)) {
     if (mode == RealtimeStoreMode::APPEND_ONLY) {
         store_write_schema_ = input_schema_;
-        store_commit_schema_ = Prepend({ValueKindField()}, input_schema_);
+        store_commit_schema_ = store_write_schema_;
         commit_schema_ = user_schema_;
         query_schema_ = Prepend({ValueKindField()}, user_schema_);
         return;

src/paimon/core/realtime/realtime_schema_layout.h

@@ -56,7 +56,7 @@ class RealtimeSchemaLayout {
         return store_write_schema_;
     }
 
-    /// Append-only fields: [_VALUE_KIND, _REALTIME_OFFSET, user fields].
+    /// Append-only fields: [_REALTIME_OFFSET, user fields].
     /// Primary-key fields: [_SEQUENCE_NUMBER, _VALUE_KIND, _REALTIME_OFFSET, user fields].
     const std::shared_ptr<arrow::Schema>& StoreCommitSchema() const {
         return store_commit_schema_;

src/paimon/core/realtime/realtime_schema_layout_test.cpp

@@ -38,8 +38,7 @@ TEST(RealtimeSchemaLayoutTest, TestSchemaLayouts) {
     ASSERT_EQ((std::vector<std::string>{"_REALTIME_OFFSET", "key", "value"}),
               append_layout->InputSchema()->field_names());
     ASSERT_TRUE(append_layout->InputSchema()->Equals(*append_layout->StoreWriteSchema()));
-    ASSERT_EQ((std::vector<std::string>{"_VALUE_KIND", "_REALTIME_OFFSET", "key", "value"}),
-              append_layout->StoreCommitSchema()->field_names());
+    ASSERT_TRUE(append_layout->StoreWriteSchema()->Equals(*append_layout->StoreCommitSchema()));
     ASSERT_TRUE(user_schema->Equals(*append_layout->CommitSchema()));
     ASSERT_EQ((std::vector<std::string>{"_VALUE_KIND", "key", "value"}),
               append_layout->QuerySchema()->field_names());