Skip to content

feat(core): support partial updates in PK realtime writes #525

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
HaHaJeff wants to merge 2 commits into
base: main
Choose a base branch
from
+301 −23
Open

feat(core): support partial updates in PK realtime writes #525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2e2f194
feat(core): support partial updates in PK realtime writes
Nov 5, 2026
3012fff
test(realtime): verify equal sequence-group updates
Nov 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions docs/source/user_guide/realtime.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down
8 changes: 4 additions & 4 deletions src/paimon/core/operation/key_value_file_store_write.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,10 +175,10 @@ Result<std::shared_ptr<BatchWriter>> KeyValueFileStoreWrite::CreateWriter(
if (!realtime_context_) {
return std::shared_ptr<BatchWriter>(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) {
Expand Down
17 changes: 11 additions & 6 deletions src/paimon/core/realtime/realtime_primary_key_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ Result<std::shared_ptr<RealtimePrimaryKeyWriter>> RealtimePrimaryKeyWriter::Crea
const std::map<std::string, std::string>& partition, int32_t bucket,
const std::shared_ptr<RealtimeSchemaLayout>& schema_layout,
const std::vector<std::string>& trimmed_primary_keys,
const std::vector<std::string>& primary_keys,
const std::shared_ptr<FieldsComparator>& key_comparator, const CoreOptions& options,
const std::shared_ptr<RealtimeContextImpl>& realtime_context,
const RealtimeStoreState& store_state, int64_t restored_max_sequence_number,
Expand All @@ -111,8 +112,10 @@ Result<std::shared_ptr<RealtimePrimaryKeyWriter>> RealtimePrimaryKeyWriter::Crea
restored_max_sequence_number == std::numeric_limits<int64_t>::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");
Expand All @@ -132,8 +135,8 @@ Result<std::shared_ptr<RealtimePrimaryKeyWriter>> RealtimePrimaryKeyWriter::Crea
partition_bucket, restored_max_sequence_number));
return std::shared_ptr<RealtimePrimaryKeyWriter>(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(
Expand All @@ -144,6 +147,7 @@ RealtimePrimaryKeyWriter::RealtimePrimaryKeyWriter(
const std::shared_ptr<RealtimeSchemaLayout>& schema_layout,
const std::shared_ptr<arrow::Schema>& key_schema,
const std::vector<std::string>& trimmed_primary_keys,
const std::vector<std::string>& primary_keys,
const std::shared_ptr<FieldsComparator>& key_comparator, const CoreOptions& options,
int64_t next_offset, int64_t last_sequence_number,
const std::shared_ptr<MemoryPool>& memory_pool)
Expand All @@ -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),
Expand Down Expand Up @@ -262,8 +267,8 @@ Status RealtimePrimaryKeyWriter::FlushSegment(
realtime_primary_key_readers) {
PAIMON_ASSIGN_OR_RAISE(
std::unique_ptr<MergeFunction> 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<MergedKeyValueRecordReader>(
std::move(realtime_primary_key_reader), key_comparator_,
std::make_shared<ReducerMergeFunctionWrapper>(std::move(merge_function))));
Expand Down
3 changes: 3 additions & 0 deletions src/paimon/core/realtime/realtime_primary_key_writer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ class RealtimePrimaryKeyWriter final : public BatchWriter {
const std::map<std::string, std::string>& partition, int32_t bucket,
const std::shared_ptr<RealtimeSchemaLayout>& schema_layout,
const std::vector<std::string>& trimmed_primary_keys,
const std::vector<std::string>& primary_keys,
const std::shared_ptr<FieldsComparator>& key_comparator, const CoreOptions& options,
const std::shared_ptr<RealtimeContextImpl>& realtime_context,
const RealtimeStoreState& store_state, int64_t restored_max_sequence_number,
Expand All @@ -77,6 +78,7 @@ class RealtimePrimaryKeyWriter final : public BatchWriter {
const std::shared_ptr<RealtimeSchemaLayout>& schema_layout,
const std::shared_ptr<arrow::Schema>& key_schema,
const std::vector<std::string>& trimmed_primary_keys,
const std::vector<std::string>& primary_keys,
const std::shared_ptr<FieldsComparator>& key_comparator,
const CoreOptions& options, int64_t next_offset,
int64_t last_sequence_number,
Expand All @@ -95,6 +97,7 @@ class RealtimePrimaryKeyWriter final : public BatchWriter {
std::vector<std::shared_ptr<RealtimeSegmentHandle>> sealed_segments_;
std::shared_ptr<arrow::Schema> key_schema_;
std::vector<std::string> trimmed_primary_keys_;
std::vector<std::string> primary_keys_;
std::shared_ptr<FieldsComparator> key_comparator_;
CoreOptions options_;
int64_t next_offset_;
Expand Down
8 changes: 5 additions & 3 deletions src/paimon/core/utils/primary_key_table_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");

@lxy-9602 lxy-9602 Nov 6, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With partial-update and DVs enabled, pushing value predicates into high-level disk files can discard rows needed to complete an in-memory partial update.

For example:

Disk (after compaction): (id=1, a="old", b="keep")
Memory partial update:  (id=1, a="new", b=null)
Query: WHERE a = 'new'

Expected: (id=1, a="new", b="keep")
Actual:   (id=1, a="new", b=null)

The disk row is filtered out before merging, so b="keep" is lost.

Please either push down only primary-key predicates across all disk levels and evaluate value predicates after merging, or reject the combination of realtime partial-update and DVs for now.

}
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()) {
Expand Down
2 changes: 1 addition & 1 deletion src/paimon/core/utils/primary_key_table_utils_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ TEST(PrimaryKeyTableUtilsTest, TestSupportedRealtimeOptions) {
TEST(PrimaryKeyTableUtilsTest, TestUnsupportedRealtimeOptions) {
const std::vector<std::map<std::string, std::string>> 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"}},
};
Expand Down
Loading
Loading