From 2e2f194ccdab76831b2988a9b749f5d8c299d8f7 Mon Sep 17 00:00:00 2001 From: "zhoujianfeng.jeff" Date: Thu, 5 Nov 2026 17:28:30 +0800 Subject: [PATCH 1/2] feat(core): support partial updates in PK realtime writes --- docs/source/user_guide/realtime.rst | 20 +- .../operation/key_value_file_store_write.cpp | 8 +- .../realtime/realtime_primary_key_writer.cpp | 17 +- .../realtime/realtime_primary_key_writer.h | 3 + .../core/utils/primary_key_table_utils.cpp | 8 +- .../utils/primary_key_table_utils_test.cpp | 2 +- test/inte/realtime_write_inte_test.cpp | 258 +++++++++++++++++- 7 files changed, 293 insertions(+), 23 deletions(-) diff --git a/docs/source/user_guide/realtime.rst b/docs/source/user_guide/realtime.rst index 5033d547a..fcf04385c 100644 --- a/docs/source/user_guide/realtime.rst +++ b/docs/source/user_guide/realtime.rst @@ -78,10 +78,11 @@ and exact residual filtering follow the mode-specific rules below. Append accepts ``INSERT`` only. PK and append deduplicate also accept ``UPDATE_AFTER``, ``DELETE``, and ``UPDATE_BEFORE``. PK supports composite -primary keys, but its real-time merge engine is currently limited to -``merge-engine=deduplicate``. Append deduplicate currently accepts exactly one -non-null, non-nested key field and is experimental: its persisted index format -may change. +primary keys with ``merge-engine=deduplicate`` or ``merge-engine=partial-update``. +Partial update uses the existing merge function and its sequence-group, +field-aggregation, and delete options. Append deduplicate currently +accepts exactly one non-null, non-nested key field and is experimental: its +persisted index format may change. PK DV and append deduplicate are different designs. PK DV retains primary-key merge semantics; append deduplicate uses a persisted stable row offset and @@ -422,11 +423,15 @@ number, row kind, and offset. Each sorted batch is an independent merge input; multiple stored batches do not form a single globally sorted stream. Query construction merges sorted disk sections and the memory readers, applies -deduplicate merge semantics, and drops final delete records. It collects all +the configured merge engine, and drops final delete records. It collects all disk splits belonging to the active partition-bucket, rather than merging memory against only one disk split. Preparing similarly merges the sealed streams before handing them to ``MergeTreeWriter`` to write normal PK files. +Partial update reuses the existing framework merge logic and options. The store +continues to hold sorted batches and spill them without merging. Real-time +compaction remains disabled. + Only primary-key predicates are safe to push into unresolved versions. A predicate on a value column must be evaluated after version merge: filtering out the newest value first could incorrectly expose an older matching value. @@ -589,8 +594,9 @@ Current restrictions include: * Process-local batch union scans only. Streaming scans, limit pushdown, global-index result splits, vector/full-text search, and count-only real-time readers are unsupported. -* PK mode supports the deduplicate merge engine with default delete handling, - ascending sequence order, and ``changelog-producer=none``. Custom +* PK mode supports deduplicate with default delete handling, or partial update + with its existing delete and sequence-group options. Both require + ascending sequence order and ``changelog-producer=none``. Custom ``sequence.field``, floating-point keys, PK global indexes, lookup without DVs, and read-optimized union scans are unsupported. * Append deduplicate requires a single non-null scalar key, a persisted offset diff --git a/src/paimon/core/operation/key_value_file_store_write.cpp b/src/paimon/core/operation/key_value_file_store_write.cpp index be72ad914..98d03dd14 100644 --- a/src/paimon/core/operation/key_value_file_store_write.cpp +++ b/src/paimon/core/operation/key_value_file_store_write.cpp @@ -175,10 +175,10 @@ Result> KeyValueFileStoreWrite::CreateWriter( if (!realtime_context_) { return std::shared_ptr(std::move(writer)); } - return RealtimePrimaryKeyWriter::Create(partition_map, bucket, realtime_schema_layout_, - trimmed_primary_keys, key_comparator_, options_, - realtime_context_impl, realtime_store_state.value(), - restore_max_seq_number, writer, pool_); + return RealtimePrimaryKeyWriter::Create( + partition_map, bucket, realtime_schema_layout_, trimmed_primary_keys, + table_schema_->PrimaryKeys(), key_comparator_, options_, realtime_context_impl, + realtime_store_state.value(), restore_max_seq_number, writer, pool_); } Status KeyValueFileStoreWrite::RefreshCommittedSnapshot(int64_t snapshot_id) { diff --git a/src/paimon/core/realtime/realtime_primary_key_writer.cpp b/src/paimon/core/realtime/realtime_primary_key_writer.cpp index 7e04b67ae..191d77bac 100644 --- a/src/paimon/core/realtime/realtime_primary_key_writer.cpp +++ b/src/paimon/core/realtime/realtime_primary_key_writer.cpp @@ -102,6 +102,7 @@ Result> RealtimePrimaryKeyWriter::Crea const std::map& partition, int32_t bucket, const std::shared_ptr& schema_layout, const std::vector& trimmed_primary_keys, + const std::vector& primary_keys, const std::shared_ptr& key_comparator, const CoreOptions& options, const std::shared_ptr& realtime_context, const RealtimeStoreState& store_state, int64_t restored_max_sequence_number, @@ -111,8 +112,10 @@ Result> RealtimePrimaryKeyWriter::Crea restored_max_sequence_number == std::numeric_limits::max()) { return Status::Invalid("PK restored sequence number is invalid"); } - if (options.GetMergeEngine() != MergeEngine::DEDUPLICATE) { - return Status::NotImplemented("PK realtime supports only the DEDUPLICATE merge engine"); + if (options.GetMergeEngine() != MergeEngine::DEDUPLICATE && + options.GetMergeEngine() != MergeEngine::PARTIAL_UPDATE) { + return Status::NotImplemented( + "PK realtime supports only the DEDUPLICATE and PARTIAL_UPDATE merge engines"); } if (!schema_layout) { return Status::Invalid("PK realtime schema layout is null"); @@ -132,8 +135,8 @@ Result> RealtimePrimaryKeyWriter::Crea partition_bucket, restored_max_sequence_number)); return std::shared_ptr(new RealtimePrimaryKeyWriter( store_state.store, merge_tree_writer, realtime_context, partition_bucket, schema_layout, - arrow::schema(std::move(key_fields)), trimmed_primary_keys, key_comparator, options, - store_state.initial_offset, initial_max_sequence_number, memory_pool)); + arrow::schema(std::move(key_fields)), trimmed_primary_keys, primary_keys, key_comparator, + options, store_state.initial_offset, initial_max_sequence_number, memory_pool)); } RealtimePrimaryKeyWriter::RealtimePrimaryKeyWriter( @@ -144,6 +147,7 @@ RealtimePrimaryKeyWriter::RealtimePrimaryKeyWriter( const std::shared_ptr& schema_layout, const std::shared_ptr& key_schema, const std::vector& trimmed_primary_keys, + const std::vector& primary_keys, const std::shared_ptr& key_comparator, const CoreOptions& options, int64_t next_offset, int64_t last_sequence_number, const std::shared_ptr& memory_pool) @@ -156,6 +160,7 @@ RealtimePrimaryKeyWriter::RealtimePrimaryKeyWriter( schema_layout_(schema_layout), key_schema_(key_schema), trimmed_primary_keys_(trimmed_primary_keys), + primary_keys_(primary_keys), key_comparator_(key_comparator), options_(options), next_offset_(next_offset), @@ -262,8 +267,8 @@ Status RealtimePrimaryKeyWriter::FlushSegment( realtime_primary_key_readers) { PAIMON_ASSIGN_OR_RAISE( std::unique_ptr merge_function, - PrimaryKeyTableUtils::CreateMergeFunction( - schema_layout_->UserSchema(), trimmed_primary_keys_, options_, memory_pool_)); + PrimaryKeyTableUtils::CreateMergeFunction(schema_layout_->UserSchema(), primary_keys_, + options_, memory_pool_)); sorted_readers.push_back(std::make_unique( std::move(realtime_primary_key_reader), key_comparator_, std::make_shared(std::move(merge_function)))); diff --git a/src/paimon/core/realtime/realtime_primary_key_writer.h b/src/paimon/core/realtime/realtime_primary_key_writer.h index 56f51bd50..77b2d7f6c 100644 --- a/src/paimon/core/realtime/realtime_primary_key_writer.h +++ b/src/paimon/core/realtime/realtime_primary_key_writer.h @@ -51,6 +51,7 @@ class RealtimePrimaryKeyWriter final : public BatchWriter { const std::map& partition, int32_t bucket, const std::shared_ptr& schema_layout, const std::vector& trimmed_primary_keys, + const std::vector& primary_keys, const std::shared_ptr& key_comparator, const CoreOptions& options, const std::shared_ptr& realtime_context, const RealtimeStoreState& store_state, int64_t restored_max_sequence_number, @@ -77,6 +78,7 @@ class RealtimePrimaryKeyWriter final : public BatchWriter { const std::shared_ptr& schema_layout, const std::shared_ptr& key_schema, const std::vector& trimmed_primary_keys, + const std::vector& primary_keys, const std::shared_ptr& key_comparator, const CoreOptions& options, int64_t next_offset, int64_t last_sequence_number, @@ -95,6 +97,7 @@ class RealtimePrimaryKeyWriter final : public BatchWriter { std::vector> sealed_segments_; std::shared_ptr key_schema_; std::vector trimmed_primary_keys_; + std::vector primary_keys_; std::shared_ptr key_comparator_; CoreOptions options_; int64_t next_offset_; diff --git a/src/paimon/core/utils/primary_key_table_utils.cpp b/src/paimon/core/utils/primary_key_table_utils.cpp index 639bb8c82..2ec8734bf 100644 --- a/src/paimon/core/utils/primary_key_table_utils.cpp +++ b/src/paimon/core/utils/primary_key_table_utils.cpp @@ -103,13 +103,15 @@ Status PrimaryKeyTableUtils::ValidateRealtimeOptions(const CoreOptions& options, if (options.GetBucket() <= 0) { return Status::NotImplemented("PK realtime requires fixed buckets"); } - if (options.GetMergeEngine() != MergeEngine::DEDUPLICATE) { - return Status::NotImplemented("PK realtime supports only the DEDUPLICATE merge engine"); + if (options.GetMergeEngine() != MergeEngine::DEDUPLICATE && + options.GetMergeEngine() != MergeEngine::PARTIAL_UPDATE) { + return Status::NotImplemented( + "PK realtime supports only the DEDUPLICATE and PARTIAL_UPDATE merge engines"); } if (options.DataEvolutionEnabled()) { return Status::NotImplemented("PK realtime does not support data evolution"); } - if (options.IgnoreDelete()) { + if (options.GetMergeEngine() == MergeEngine::DEDUPLICATE && options.IgnoreDelete()) { return Status::NotImplemented("PK realtime requires default delete behavior"); } if (!options.GetSequenceField().empty()) { diff --git a/src/paimon/core/utils/primary_key_table_utils_test.cpp b/src/paimon/core/utils/primary_key_table_utils_test.cpp index 5f8a7cb3a..a1fa773ea 100644 --- a/src/paimon/core/utils/primary_key_table_utils_test.cpp +++ b/src/paimon/core/utils/primary_key_table_utils_test.cpp @@ -67,7 +67,7 @@ TEST(PrimaryKeyTableUtilsTest, TestSupportedRealtimeOptions) { TEST(PrimaryKeyTableUtilsTest, TestUnsupportedRealtimeOptions) { const std::vector> unsupported_options = { {{Options::BUCKET, "0"}}, - {{Options::BUCKET, "1"}, {Options::MERGE_ENGINE, "partial-update"}}, + {{Options::BUCKET, "1"}, {Options::MERGE_ENGINE, "aggregation"}}, {{Options::BUCKET, "1"}, {Options::DATA_EVOLUTION_ENABLED, "true"}}, {{Options::BUCKET, "1"}, {Options::SEQUENCE_FIELD, "seq"}}, }; diff --git a/test/inte/realtime_write_inte_test.cpp b/test/inte/realtime_write_inte_test.cpp index e520d43e9..38f14d0aa 100644 --- a/test/inte/realtime_write_inte_test.cpp +++ b/test/inte/realtime_write_inte_test.cpp @@ -716,7 +716,7 @@ class RealtimeWriteInteTest : public ::testing::Test { } Result> MakeUnpartitionedBatchFromJson( - const std::string& json) const { + const std::string& json, const std::vector& row_kinds = {}) const { PAIMON_ASSIGN_OR_RAISE( std::unique_ptr schema_layout, RealtimeSchemaLayout::Create(RealtimeStoreMode::APPEND_ONLY, schema_)); @@ -726,7 +726,7 @@ class RealtimeWriteInteTest : public ::testing::Test { arrow::struct_(realtime_schema->fields()), json)); ArrowArray c_array; PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array)); - return RecordBatchBuilder(&c_array).SetBucket(0).Finish(); + return RecordBatchBuilder(&c_array).SetBucket(0).SetRowKinds(row_kinds).Finish(); } static std::vector MakeRows(int64_t first_id, int64_t count, @@ -992,6 +992,10 @@ class RealtimeWriteInteTest : public ::testing::Test { // Match the dictionary and nullability normalization applied by ReadResultCollector. ASSERT_OK_AND_ASSIGN( expected, DictArrayConverter::ConvertDictArray(expected, arrow::default_memory_pool())); + if (expected->length() == 0) { + ASSERT_TRUE(!result || result->length() == 0); + return; + } ASSERT_NE(nullptr, result); ASSERT_TRUE(std::make_shared(expected)->Equals(*result)) << result->ToString(); @@ -4321,6 +4325,256 @@ TEST_F(RealtimeWriteInteTest, TestPkDeleteInsertAndPinnedReadsAcrossRefresh) { ASSERT_OK(writer->Close()); } +TEST_F(RealtimeWriteInteTest, TestPkPartialUpdateDeleteAcrossCommits) { + fields_ = {arrow::field("id", arrow::int64()), arrow::field("a", arrow::int64()), + arrow::field("b", arrow::int64())}; + schema_ = arrow::schema(fields_); + options_[Options::FILE_FORMAT] = "parquet"; + options_[Options::MERGE_ENGINE] = "partial-update"; + options_[Options::PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE] = "true"; + options_[Options::READ_BATCH_SIZE] = "1"; + CreatePkTable(); + // Delete and reinsert use separate commits; same-commit delete boundaries are not covered. + const std::vector inputs = {"[[0, 1, 10, 20]]", "[[1, 1, 30, null]]", + "[[2, 1, null, null]]", "[[3, 1, 50, null]]", + "[[4, 1, null, 40]]"}; + const std::vector kinds = { + RecordBatch::RowKind::INSERT, RecordBatch::RowKind::UPDATE_AFTER, + RecordBatch::RowKind::DELETE, RecordBatch::RowKind::UPDATE_AFTER, + RecordBatch::RowKind::INSERT}; + const std::vector expected = {"[[0, 1, 10, 20]]", "[[0, 1, 30, 20]]", "[]", + "[[0, 1, 50, null]]", "[[0, 1, 50, 40]]"}; + for (int64_t commit = 0; commit < static_cast(inputs.size()); ++commit) { + SCOPED_TRACE(commit); + ASSERT_OK_AND_ASSIGN(std::shared_ptr context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, CreateRealtimeWriter(context)); + if (commit > 0) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, expected[commit - 1])); + } + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, + MakeUnpartitionedBatchFromJson(inputs[commit], {kinds[commit]})); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, expected[commit])); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(commit)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, expected[commit])); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(progress, commit)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, expected[commit])); + ASSERT_OK(writer->Close()); + } +} + +TEST_F(RealtimeWriteInteTest, TestPkPartialUpdateSequenceGroups) { + fields_ = {arrow::field("id", arrow::int64()), arrow::field("sa", arrow::int64()), + arrow::field("a", arrow::int64()), arrow::field("sb", arrow::int64()), + arrow::field("b", arrow::int64()), arrow::field("note", arrow::utf8())}; + schema_ = arrow::schema(fields_); + options_[Options::FILE_FORMAT] = "parquet"; + options_[Options::MERGE_ENGINE] = "partial-update"; + options_["fields.sa.sequence-group"] = "a"; + options_["fields.sb.sequence-group"] = "b"; + options_[Options::READ_BATCH_SIZE] = "2"; + CreatePkTable(); + ASSERT_OK_AND_ASSIGN(std::shared_ptr context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, CreateRealtimeWriter(context)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, MakeUnpartitionedBatchFromJson(R"([ + [0, 1, 10, 100, 20, 200, "seed"], [1, 2, 10, 10, 20, 20, "two"] + ])")); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(0)); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(progress, 0)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + + // Older, equal and null group versions are independent; ungrouped nulls keep prior values. + ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(R"([ + [2, 1, 9, 999, 21, 210, null], + [3, 1, 10, 110, null, 999, null], + [4, 1, 11, null, 19, 190, null] + ])")); + ASSERT_OK(writer->Write(std::move(batch))); + const std::string expected = + R"([[0, 1, 11, null, 21, 210, "seed"], [0, 2, 10, 10, 20, 20, "two"]])"; + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck(plan, context, schema_, expected)); + ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(1)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck(plan, context, schema_, expected)); + ASSERT_OK_AND_ASSIGN(snapshot, Commit(progress, 1)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, expected)); + ASSERT_OK(writer->Close()); + writer.reset(); + plan.reset(); + context.reset(); + + ASSERT_OK_AND_ASSIGN(context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(writer, CreateRealtimeWriter(context)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck(plan, context, schema_, expected)); + ASSERT_OK_AND_ASSIGN(batch, + MakeUnpartitionedBatchFromJson(R"([[5, 1, 12, 120, null, 999, null]])")); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(2)); + ASSERT_OK_AND_ASSIGN(snapshot, Commit(progress, 2)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck( + plan, nullptr, schema_, + R"([[0, 1, 12, 120, 21, 210, "seed"], [0, 2, 10, 10, 20, 20, "two"]])")); + ASSERT_OK(writer->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestPkPartialUpdateAggregationSpillAndRecovery) { + fields_ = {arrow::field("id", arrow::int64()), arrow::field("seq", arrow::int64()), + arrow::field("total", arrow::int64()), arrow::field("last", arrow::utf8())}; + schema_ = arrow::schema(fields_); + options_[Options::FILE_FORMAT] = "parquet"; + options_[Options::MERGE_ENGINE] = "partial-update"; + options_["fields.seq.sequence-group"] = "total,last"; + options_["fields.total.aggregate-function"] = "sum"; + options_["fields.last.aggregate-function"] = "last_non_null_value"; + options_[Options::REALTIME_SPILL_ENABLED] = "true"; + options_[Options::READ_BATCH_SIZE] = "1"; + CreatePkTable(); + const std::string temp_directory = PathUtil::JoinPath(dir_->Str(), "partial-update-spill"); + ASSERT_OK_AND_ASSIGN(std::shared_ptr context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, + CreateRealtimeWriter(context, temp_directory)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr batch, MakeUnpartitionedBatchFromJson(R"([ + [0, 1, 9, 6, "nine"], [1, 1, 10, 4, "ten"] + ])")); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 10, 10, "ten"]])")); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(0)); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(progress, 0)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, R"([[0, 1, 10, 10, "ten"]])")); + ASSERT_OK_AND_ASSIGN(int64_t count, CountRows(plan, nullptr)); + ASSERT_EQ(1, count); + + // Keep the retract with its contribution; no all-retract batch or complex last provenance. + ASSERT_OK_AND_ASSIGN(batch, + MakeUnpartitionedBatchFromJson( + R"([[2, 1, 20, 20, "twenty"], [3, 1, 21, 3, null]])", + {RecordBatch::RowKind::INSERT, RecordBatch::RowKind::UPDATE_BEFORE})); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK(writer->Seal()); + ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(), temp_directory)); + ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(R"([[4, 1, 22, 5, "latest"]])")); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 22, 32, "latest"]])")); + ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(1)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 22, 32, "latest"]])")); + ASSERT_OK_AND_ASSIGN(snapshot, Commit(progress, 1)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_TRUE(WaitForChannelFileCount(temp_directory, 0)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, R"([[0, 1, 22, 32, "latest"]])")); + const std::string tail = R"([[5, 1, 23, 7, "tail"]])"; + ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(tail)); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 23, 39, "tail"]])")); + ASSERT_NOK_WITH_MSG(writer->Close(), + "real-time writer closed with data not covered by a successful " + "PrepareCommitWithProgress"); + writer.reset(); + plan.reset(); + context.reset(); + + ASSERT_OK_AND_ASSIGN(context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(writer, CreateRealtimeWriter(context, temp_directory)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 22, 32, "latest"]])")); + // Simulate caller WAL replay with exactly the original offset and row. + ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(tail)); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 23, 39, "tail"]])")); + ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(2)); + ASSERT_OK_AND_ASSIGN(snapshot, Commit(progress, 2)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, R"([[0, 1, 23, 39, "tail"]])")); + ASSERT_OK(writer->Close()); +} + +TEST_F(RealtimeWriteInteTest, TestPkPartialUpdatePartitionedDefaultAggregationRetract) { + fields_ = {arrow::field("id", arrow::int64()), arrow::field("pt", arrow::utf8()), + arrow::field("seq", arrow::int64()), arrow::field("total", arrow::int64())}; + schema_ = arrow::schema(fields_); + options_[Options::FILE_FORMAT] = "parquet"; + options_[Options::MERGE_ENGINE] = "partial-update"; + options_[Options::FIELDS_DEFAULT_AGG_FUNC] = "sum"; + options_["fields.seq.sequence-group"] = "total"; + CreatePkTable({"pt"}); + ASSERT_OK_AND_ASSIGN(std::shared_ptr context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, CreateRealtimeWriter(context)); + ASSERT_OK_AND_ASSIGN(std::unique_ptr layout, + RealtimeSchemaLayout::Create(RealtimeStoreMode::APPEND_ONLY, schema_)); + auto array_result = + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(layout->InputSchema()->fields()), + R"([[0, 1, "p0", 10, 10], [1, 1, "p0", 11, 3]])"); + ASSERT_TRUE(array_result.ok()); + std::shared_ptr array = array_result.ValueOrDie(); + ArrowArray c_array; + ASSERT_TRUE(arrow::ExportArray(*array, &c_array).ok()); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr batch, + RecordBatchBuilder(&c_array) + .SetPartition({{"pt", "p0"}}) + .SetBucket(0) + .SetRowKinds({RecordBatch::RowKind::INSERT, RecordBatch::RowKind::DELETE}) + .Finish()); + ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, "p0", 11, 7]])")); + ASSERT_OK_AND_ASSIGN(std::vector progress, + writer->PrepareCommitWithProgress(0)); + ASSERT_OK_AND_ASSIGN(int64_t snapshot, Commit(progress, 0)); + ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, R"([[0, 1, "p0", 11, 7]])")); + ASSERT_OK(writer->Close()); + writer.reset(); + plan.reset(); + context.reset(); + + ASSERT_OK_AND_ASSIGN(context, RealtimeContext::Create()); + ASSERT_OK_AND_ASSIGN(writer, CreateRealtimeWriter(context)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE( + ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, "p0", 11, 7]])")); + ASSERT_OK(writer->Close()); +} + TEST_F(RealtimeWriteInteTest, TestPkMergeDiskSealedAndActive) { options_[Options::READ_BATCH_SIZE] = "2"; CreatePkTable(); From 3012fff9e0c03b02cc74b68497c22ab24bab0390 Mon Sep 17 00:00:00 2001 From: "zhoujianfeng.jeff" Date: Fri, 6 Nov 2026 16:30:26 +0800 Subject: [PATCH 2/2] test(realtime): verify equal sequence-group updates --- test/inte/realtime_write_inte_test.cpp | 28 +++++++++++++++++--------- 1 file changed, 18 insertions(+), 10 deletions(-) diff --git a/test/inte/realtime_write_inte_test.cpp b/test/inte/realtime_write_inte_test.cpp index 38f14d0aa..ab57ab924 100644 --- a/test/inte/realtime_write_inte_test.cpp +++ b/test/inte/realtime_write_inte_test.cpp @@ -4398,13 +4398,19 @@ TEST_F(RealtimeWriteInteTest, TestPkPartialUpdateSequenceGroups) { // Older, equal and null group versions are independent; ungrouped nulls keep prior values. ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(R"([ [2, 1, 9, 999, 21, 210, null], - [3, 1, 10, 110, null, 999, null], - [4, 1, 11, null, 19, 190, null] + [3, 1, 10, 110, null, 999, null] ])")); ASSERT_OK(writer->Write(std::move(batch))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck( + plan, context, schema_, + R"([[0, 1, 10, 110, 21, 210, "seed"], [0, 2, 10, 10, 20, 20, "two"]])")); + ASSERT_OK_AND_ASSIGN(batch, + MakeUnpartitionedBatchFromJson(R"([[4, 1, 11, null, 19, 190, null]])")); + ASSERT_OK(writer->Write(std::move(batch))); const std::string expected = R"([[0, 1, 11, null, 21, 210, "seed"], [0, 2, 10, 10, 20, 20, "two"]])"; - ASSERT_OK_AND_ASSIGN(std::shared_ptr plan, CreatePlan(context, nullptr)); + ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); ASSERT_NO_FATAL_FAILURE(ReadPlanWithSchemaAndCheck(plan, context, schema_, expected)); ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(1)); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); @@ -4478,25 +4484,27 @@ TEST_F(RealtimeWriteInteTest, TestPkPartialUpdateAggregationSpillAndRecovery) { ASSERT_EQ(1, TestHelper::CountChannelFiles(dir_->GetFileSystem(), temp_directory)); ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(R"([[4, 1, 22, 5, "latest"]])")); ASSERT_OK(writer->Write(std::move(batch))); + const std::string committed_expected = R"([[0, 1, 22, 32, "latest"]])"; + const std::string replayed_expected = R"([[0, 1, 23, 39, "tail"]])"; ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 22, 32, "latest"]])")); + ReadPlanWithSchemaAndCheck(plan, context, schema_, committed_expected)); ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(1)); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 22, 32, "latest"]])")); + ReadPlanWithSchemaAndCheck(plan, context, schema_, committed_expected)); ASSERT_OK_AND_ASSIGN(snapshot, Commit(progress, 1)); ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); ASSERT_TRUE(WaitForChannelFileCount(temp_directory, 0)); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, R"([[0, 1, 22, 32, "latest"]])")); + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, committed_expected)); const std::string tail = R"([[5, 1, 23, 7, "tail"]])"; ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(tail)); ASSERT_OK(writer->Write(std::move(batch))); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 23, 39, "tail"]])")); + ReadPlanWithSchemaAndCheck(plan, context, schema_, replayed_expected)); ASSERT_NOK_WITH_MSG(writer->Close(), "real-time writer closed with data not covered by a successful " "PrepareCommitWithProgress"); @@ -4508,19 +4516,19 @@ TEST_F(RealtimeWriteInteTest, TestPkPartialUpdateAggregationSpillAndRecovery) { ASSERT_OK_AND_ASSIGN(writer, CreateRealtimeWriter(context, temp_directory)); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 22, 32, "latest"]])")); + ReadPlanWithSchemaAndCheck(plan, context, schema_, committed_expected)); // Simulate caller WAL replay with exactly the original offset and row. ASSERT_OK_AND_ASSIGN(batch, MakeUnpartitionedBatchFromJson(tail)); ASSERT_OK(writer->Write(std::move(batch))); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(context, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, context, schema_, R"([[0, 1, 23, 39, "tail"]])")); + ReadPlanWithSchemaAndCheck(plan, context, schema_, replayed_expected)); ASSERT_OK_AND_ASSIGN(progress, writer->PrepareCommitWithProgress(2)); ASSERT_OK_AND_ASSIGN(snapshot, Commit(progress, 2)); ASSERT_OK(writer->RefreshCommittedSnapshot(snapshot)); ASSERT_OK_AND_ASSIGN(plan, CreatePlan(nullptr, nullptr)); ASSERT_NO_FATAL_FAILURE( - ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, R"([[0, 1, 23, 39, "tail"]])")); + ReadPlanWithSchemaAndCheck(plan, nullptr, schema_, replayed_expected)); ASSERT_OK(writer->Close()); }