From f11129718015911f5667746532395ef097cd5a9f Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Thu, 27 Aug 2026 04:29:32 -0400 Subject: [PATCH 1/3] fix(index): support partially active PK index groups Keep the full immutable source metadata for ordinal localization while covering only eligible sources present in the current scan. Reject conflicting or mismatched active intersections and leave uncovered files on the normal scan path. --- .../user_guide/primary_key_global_index.rst | 16 +- .../pksorted/pk_sorted_bucket_index_state.cpp | 115 ++++++++-- .../pksorted/pk_sorted_bucket_index_state.h | 11 +- .../pk_sorted_bucket_index_state_test.cpp | 150 ++++++++++-- .../index/pksorted/pk_sorted_index_group.h | 14 +- .../source/primary_key_sorted_index_scan.cpp | 5 +- .../primary_key_sorted_index_scan_test.cpp | 217 ++++++++++++++++++ 7 files changed, 465 insertions(+), 63 deletions(-) diff --git a/docs/source/user_guide/primary_key_global_index.rst b/docs/source/user_guide/primary_key_global_index.rst index 3b9aaa56e..8964ee5a1 100644 --- a/docs/source/user_guide/primary_key_global_index.rst +++ b/docs/source/user_guide/primary_key_global_index.rst @@ -21,8 +21,8 @@ Primary Key Global Index Paimon 2.0 primary-key tables support *source-backed* global scalar indexes (``pk-btree`` / ``pk-bitmap``). Unlike the Data Evolution global indexes described in :doc:`global_index`, which address rows by a table-wide row id, a source-backed payload -covers the complete active source set of one positive data level of one bucket, and its -results are group ordinals that are localized back to per-file physical row positions. +covers an immutable ordered source group from one positive data level of one bucket, and +its results are group ordinals that are localized back to per-file physical row positions. paimon-cpp supports the read path of this protocol: ordinary batch scans of a primary-key table with scalar index definitions automatically evaluate the part of the @@ -43,11 +43,13 @@ The definitions follow the Java table options: Semantics --------- -- A payload is only used when it provably covers the current active source set of its - data level: exactly one payload per level, source file names / order / row counts - identical to the active COMPACT files of that level, matching index type and field id, - and a row range of exactly ``[0, total source rows - 1]``. Anything else is treated as - uncovered and scanned normally. +- A payload retains its complete ordered source list as the group-ordinal namespace. If + part of that list is not in the current scan because it was retired or safely pruned, + the payload can still cover the remaining files. Several payloads may coexist when their + active source intersections are disjoint. A payload is rejected if it has no active + source, an active source's row count differs, its metadata or row range is invalid, or + another payload claims the same active source. Active files without accepted coverage + are scanned normally. - ``AND`` predicates narrow with any safely evaluable indexed child; ``OR`` predicates only use the index when every branch is evaluable. Files whose evaluation fails, whose positions are out of range, or whose result needs more than 4096 ranges fall back to a diff --git a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp index 7d3b5a6d1..ca2e5dc9a 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp +++ b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include "paimon/core/index/global_index_meta.h" @@ -30,6 +31,15 @@ #include "paimon/core/index/pk/primary_key_index_source_policy.h" namespace paimon { +namespace { +struct PayloadCandidate { + std::shared_ptr payload; + std::shared_ptr group; + std::vector active_sources; + bool conflicted = false; +}; +} // namespace + PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles( int32_t field_id, const std::string& index_type, const std::vector>& active_data_files, @@ -49,10 +59,20 @@ PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles( }); } - // Match payloads against the expected level sources; anything that does not decode or - // does not exactly cover its level is rejected. - std::map>> payloads_by_level; - std::map> payload_metas_by_level; + std::map active_sources; + std::set ambiguous_active_source_names; + for (const auto& level_sources : sources_by_level) { + for (const PrimaryKeyIndexSourceFile& source : level_sources.second) { + if (!active_sources.emplace(source.file_name, source.row_count).second) { + ambiguous_active_source_names.insert(source.file_name); + } + } + } + + // Keep the complete immutable source group for ordinal localization, but only claim the + // sources which are still active in this snapshot. Several disjoint groups may coexist + // after an update; no active source file may be claimed by two groups. + std::vector candidates; std::vector> rejected; for (const std::shared_ptr& payload : active_payloads) { if (payload == nullptr) { @@ -71,39 +91,86 @@ PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles( continue; } PrimaryKeyIndexSourceMeta source_meta = std::move(source_meta_result).value(); - auto desired = sources_by_level.find(source_meta.DataLevel()); - if (desired == sources_by_level.end() || desired->second != source_meta.SourceFiles()) { + const std::vector& payload_sources = source_meta.SourceFiles(); + bool valid_candidate = !payload_sources.empty(); + std::vector active_intersection; + for (size_t i = 0; valid_candidate && i < payload_sources.size(); i++) { + const PrimaryKeyIndexSourceFile& source = payload_sources[i]; + if (i > 0 && payload_sources[i - 1].file_name >= source.file_name) { + valid_candidate = false; + break; + } + auto active_source = active_sources.find(source.file_name); + if (active_source == active_sources.end()) { + continue; + } + if (active_source->second != source.row_count || + ambiguous_active_source_names.count(source.file_name) > 0) { + valid_candidate = false; + break; + } + active_intersection.push_back(source); + } + if (!valid_candidate || active_intersection.empty()) { + rejected.push_back(payload); + continue; + } + std::shared_ptr group = + PkSortedIndexGroup::Create(field_id, index_type, payload_sources, payload, source_meta); + if (group == nullptr) { rejected.push_back(payload); continue; } - payloads_by_level[source_meta.DataLevel()].push_back(payload); - payload_metas_by_level[source_meta.DataLevel()].push_back(std::move(source_meta)); + candidates.push_back( + {payload, std::move(group), std::move(active_intersection), /*conflicted=*/false}); + } + + std::map, std::vector> candidates_by_source; + for (size_t i = 0; i < candidates.size(); i++) { + const PayloadCandidate& candidate = candidates[i]; + for (const PrimaryKeyIndexSourceFile& source : candidate.active_sources) { + candidates_by_source[{source.file_name, source.row_count}].push_back(i); + } + } + for (const auto& source_candidates : candidates_by_source) { + if (source_candidates.second.size() > 1) { + for (size_t candidate_index : source_candidates.second) { + candidates[candidate_index].conflicted = true; + } + } } std::vector> groups; - std::set covered_levels; - for (const auto& level_payloads : payloads_by_level) { - int32_t level = level_payloads.first; - std::shared_ptr group; - if (level_payloads.second.size() == 1) { - group = PkSortedIndexGroup::Create(field_id, index_type, sources_by_level[level], - level_payloads.second[0], - payload_metas_by_level[level][0]); + std::set> covered_sources; + for (PayloadCandidate& candidate : candidates) { + if (candidate.conflicted) { + rejected.push_back(std::move(candidate.payload)); + continue; } - if (group != nullptr) { - groups.push_back(std::move(group)); - covered_levels.insert(level); - } else { - rejected.insert(rejected.end(), level_payloads.second.begin(), - level_payloads.second.end()); + for (const PrimaryKeyIndexSourceFile& source : candidate.active_sources) { + covered_sources.emplace(source.file_name, source.row_count); } + groups.push_back(std::move(candidate.group)); } + std::sort(groups.begin(), groups.end(), + [](const std::shared_ptr& left, + const std::shared_ptr& right) { + if (left->DataLevel() != right->DataLevel()) { + return left->DataLevel() < right->DataLevel(); + } + return left->SourceFiles().front().file_name < + right->SourceFiles().front().file_name; + }); std::vector covered; std::vector uncovered; for (const auto& level_sources : sources_by_level) { - auto& target = covered_levels.count(level_sources.first) > 0 ? covered : uncovered; - target.insert(target.end(), level_sources.second.begin(), level_sources.second.end()); + for (const PrimaryKeyIndexSourceFile& source : level_sources.second) { + auto& target = covered_sources.count({source.file_name, source.row_count}) > 0 + ? covered + : uncovered; + target.push_back(source); + } } return PkSortedBucketIndexState(std::move(groups), std::move(covered), std::move(uncovered), std::move(rejected)); diff --git a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h index 6923010da..89474d015 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h +++ b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h @@ -33,10 +33,13 @@ namespace paimon { /// Immutable source-backed sorted-index state for one field and bucket. /// -/// Derives the eligible per-level source sets from the active data files, matches the -/// active payloads against them, and keeps the exact validated groups. Payloads whose -/// source metadata cannot be decoded or does not exactly cover its level are rejected; -/// levels without a valid group stay uncovered and must be scanned normally. +/// Derives the eligible source sets from the active data files and keeps validated, +/// non-overlapping payload groups. A group retains its complete immutable source list for +/// ordinal localization while covering only the listed files which are still active. A +/// payload is rejected when its metadata is invalid, it has no active source, an active +/// source has a different row count, its source order is not canonical, or it overlaps +/// another candidate on an active source. Active files without an accepted group remain +/// uncovered and must be scanned normally. class PkSortedBucketIndexState { public: static PkSortedBucketIndexState FromActiveDataFiles( diff --git a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp index ec65f4ff1..06f93dbe0 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp +++ b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp @@ -57,18 +57,35 @@ class PkSortedBucketIndexStateTest : public ::testing::Test { } /// Builds a payload whose source metadata lists the given sources in the given order. - std::shared_ptr MakePayload( - int32_t field_id, const std::string& index_type, int32_t data_level, - const std::vector& sources, int64_t total_row_count, - int64_t row_range_start, int64_t row_range_end) const { + std::shared_ptr MakeNamedPayload( + const std::string& payload_name, int32_t field_id, const std::string& index_type, + int32_t data_level, const std::vector& sources, + int64_t total_row_count, int64_t row_range_start, int64_t row_range_end) const { EXPECT_OK_AND_ASSIGN(PrimaryKeyIndexSourceMeta source_meta, PrimaryKeyIndexSourceMeta::Create(data_level, sources)); EXPECT_OK_AND_ASSIGN(std::shared_ptr source_meta_bytes, source_meta.Serialize(pool_)); - return MakePayloadWithSourceMetaBytes(field_id, index_type, total_row_count, + return MakePayloadWithSourceMetaBytes(payload_name, field_id, index_type, total_row_count, row_range_start, row_range_end, source_meta_bytes); } + std::shared_ptr MakePayload( + int32_t field_id, const std::string& index_type, int32_t data_level, + const std::vector& sources, int64_t total_row_count, + int64_t row_range_start, int64_t row_range_end) const { + return MakeNamedPayload("payload.index", field_id, index_type, data_level, sources, + total_row_count, row_range_start, row_range_end); + } + + std::shared_ptr MakeNamedPayload( + const std::string& payload_name, int32_t field_id, const std::string& index_type, + int32_t data_level, const std::vector& sources, + int64_t total_row_count) const { + return MakeNamedPayload(payload_name, field_id, index_type, data_level, sources, + total_row_count, /*row_range_start=*/0, + /*row_range_end=*/total_row_count - 1); + } + std::shared_ptr MakePayload( int32_t field_id, const std::string& index_type, int32_t data_level, const std::vector& sources, int64_t total_row_count) const { @@ -77,13 +94,13 @@ class PkSortedBucketIndexStateTest : public ::testing::Test { } std::shared_ptr MakePayloadWithSourceMetaBytes( - int32_t field_id, const std::string& index_type, int64_t total_row_count, - int64_t row_range_start, int64_t row_range_end, + const std::string& payload_name, int32_t field_id, const std::string& index_type, + int64_t total_row_count, int64_t row_range_start, int64_t row_range_end, const std::shared_ptr& source_meta_bytes) const { GlobalIndexMeta global_index_meta(row_range_start, row_range_end, field_id, /*extra_field_ids=*/std::nullopt, /*index_meta=*/nullptr, source_meta_bytes); - return std::make_shared(index_type, /*file_name=*/"payload.index", + return std::make_shared(index_type, payload_name, /*file_size=*/2048, total_row_count, /*dv_ranges=*/std::nullopt, /*external_path=*/std::nullopt, global_index_meta); @@ -159,23 +176,116 @@ TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWithMismatchedSourceRowCount) ASSERT_EQ(2, state.UncoveredSourceFiles().size()); } -TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadsCoveringWrongSourceSet) { +TEST_F(PkSortedBucketIndexStateTest, AcceptsActiveSubsetAndLeavesOtherFilesUncovered) { std::vector> data_files = { MakeDataFile("a", 100, 5, FileSource::Compact()), MakeDataFile("b", 200, 5, FileSource::Compact())}; - std::shared_ptr missing_source_payload = - MakePayload(7, "btree", 5, {{"a", 100}}, 100); - std::shared_ptr extra_source_payload = - MakePayload(7, "btree", 5, {{"a", 100}, {"b", 200}, {"c", 50}}, 350); + std::shared_ptr subset_payload = MakePayload(7, "btree", 5, {{"a", 100}}, 100); + PkSortedBucketIndexState state = + PkSortedBucketIndexState::FromActiveDataFiles(7, "btree", data_files, {subset_payload}); + ASSERT_EQ(1, state.Groups().size()); + ASSERT_EQ(subset_payload, state.Groups()[0]->Payload()); + ASSERT_EQ((std::vector{{"a", 100}}), state.CoveredSourceFiles()); + ASSERT_EQ((std::vector{{"b", 200}}), state.UncoveredSourceFiles()); + ASSERT_TRUE(state.RejectedPayloads().empty()); +} + +TEST_F(PkSortedBucketIndexStateTest, AcceptsDisjointPayloadGroupsAtSameLevel) { + std::vector> data_files = { + MakeDataFile("a", 100, 5, FileSource::Compact()), + MakeDataFile("b", 200, 5, FileSource::Compact()), + MakeDataFile("c", 50, 5, FileSource::Compact())}; + std::shared_ptr first_payload = + MakeNamedPayload("first.index", 7, "btree", 5, {{"a", 100}, {"b", 200}}, 300); + std::shared_ptr second_payload = + MakeNamedPayload("second.index", 7, "btree", 5, {{"c", 50}}, 50); PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( - 7, "btree", data_files, {missing_source_payload, extra_source_payload}); + 7, "btree", data_files, {second_payload, first_payload}); + ASSERT_EQ(2, state.Groups().size()); + ASSERT_EQ(first_payload, state.Groups()[0]->Payload()); + ASSERT_EQ(second_payload, state.Groups()[1]->Payload()); + ASSERT_EQ((std::vector{{"a", 100}, {"b", 200}, {"c", 50}}), + state.CoveredSourceFiles()); + ASSERT_TRUE(state.UncoveredSourceFiles().empty()); + ASSERT_TRUE(state.RejectedPayloads().empty()); +} + +TEST_F(PkSortedBucketIndexStateTest, RetainsRetiredSourcesButCoversOnlyActiveIntersection) { + std::vector> data_files = { + MakeDataFile("a", 100, 5, FileSource::Compact()), + MakeDataFile("b", 200, 5, FileSource::Compact())}; + std::shared_ptr mixed_payload = + MakeNamedPayload("mixed.index", 7, "btree", 5, {{"a", 100}, {"retired", 50}}, 150); + std::shared_ptr valid_payload = + MakeNamedPayload("valid.index", 7, "btree", 5, {{"b", 200}}, 200); + PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( + 7, "btree", data_files, {mixed_payload, valid_payload}); + ASSERT_EQ(2, state.Groups().size()); + ASSERT_EQ(mixed_payload, state.Groups()[0]->Payload()); + ASSERT_EQ((std::vector{{"a", 100}, {"retired", 50}}), + state.Groups()[0]->SourceFiles()); + ASSERT_EQ(valid_payload, state.Groups()[1]->Payload()); + ASSERT_EQ((std::vector{{"a", 100}, {"b", 200}}), + state.CoveredSourceFiles()); + ASSERT_TRUE(state.UncoveredSourceFiles().empty()); + ASSERT_TRUE(state.RejectedPayloads().empty()); +} + +TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWhoseSourcesAreAllRetired) { + std::vector> data_files = { + MakeDataFile("active", 100, 5, FileSource::Compact())}; + std::shared_ptr retired_payload = + MakePayload(7, "btree", 5, {{"retired", 50}}, 50); + PkSortedBucketIndexState state = + PkSortedBucketIndexState::FromActiveDataFiles(7, "btree", data_files, {retired_payload}); ASSERT_TRUE(state.Groups().empty()); - ASSERT_EQ(2, state.RejectedPayloads().size()); ASSERT_TRUE(state.CoveredSourceFiles().empty()); - ASSERT_EQ(2, state.UncoveredSourceFiles().size()); + ASSERT_EQ((std::vector{{"active", 100}}), + state.UncoveredSourceFiles()); + ASSERT_EQ((std::vector>{retired_payload}), + state.RejectedPayloads()); } -TEST_F(PkSortedBucketIndexStateTest, RejectsBothPayloadsWhenLevelHasTwoCandidates) { +TEST_F(PkSortedBucketIndexStateTest, RetiredSourceOverlapDoesNotConflict) { + std::vector> data_files = { + MakeDataFile("b-active", 100, 5, FileSource::Compact()), + MakeDataFile("c-active", 200, 5, FileSource::Compact())}; + std::shared_ptr left_payload = + MakeNamedPayload("left.index", 7, "btree", 5, {{"a-retired", 50}, {"b-active", 100}}, 150); + std::shared_ptr right_payload = + MakeNamedPayload("right.index", 7, "btree", 5, {{"a-retired", 50}, {"c-active", 200}}, 250); + PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( + 7, "btree", data_files, {left_payload, right_payload}); + ASSERT_EQ(2, state.Groups().size()); + ASSERT_EQ((std::vector{{"b-active", 100}, {"c-active", 200}}), + state.CoveredSourceFiles()); + ASSERT_TRUE(state.UncoveredSourceFiles().empty()); + ASSERT_TRUE(state.RejectedPayloads().empty()); +} + +TEST_F(PkSortedBucketIndexStateTest, RejectsOverlappingGroupsButKeepsDisjointGroup) { + std::vector> data_files = { + MakeDataFile("a", 100, 5, FileSource::Compact()), + MakeDataFile("b", 200, 5, FileSource::Compact()), + MakeDataFile("c", 50, 5, FileSource::Compact()), + MakeDataFile("d", 75, 5, FileSource::Compact())}; + std::shared_ptr left_payload = + MakeNamedPayload("left.index", 7, "btree", 5, {{"a", 100}, {"b", 200}}, 300); + std::shared_ptr right_payload = + MakeNamedPayload("right.index", 7, "btree", 5, {{"b", 200}, {"c", 50}}, 250); + std::shared_ptr disjoint_payload = + MakeNamedPayload("disjoint.index", 7, "btree", 5, {{"d", 75}}, 75); + PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( + 7, "btree", data_files, {left_payload, disjoint_payload, right_payload}); + ASSERT_EQ(1, state.Groups().size()); + ASSERT_EQ(disjoint_payload, state.Groups()[0]->Payload()); + ASSERT_EQ((std::vector{{"d", 75}}), state.CoveredSourceFiles()); + ASSERT_EQ((std::vector{{"a", 100}, {"b", 200}, {"c", 50}}), + state.UncoveredSourceFiles()); + ASSERT_EQ(2, state.RejectedPayloads().size()); +} + +TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadsClaimingTheSameActiveSources) { std::vector> data_files = { MakeDataFile("a", 100, 5, FileSource::Compact()), MakeDataFile("b", 200, 5, FileSource::Compact())}; @@ -268,9 +378,9 @@ TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWithCorruptSourceMeta) { // Version 1 followed by a truncated data level. std::shared_ptr corrupt_source_meta = std::make_shared(std::string("\x00\x00\x00\x01\x00\x00", 6), pool_.get()); - std::shared_ptr payload = - MakePayloadWithSourceMetaBytes(7, "btree", /*total_row_count=*/300, /*row_range_start=*/0, - /*row_range_end=*/299, corrupt_source_meta); + std::shared_ptr payload = MakePayloadWithSourceMetaBytes( + "payload.index", 7, "btree", /*total_row_count=*/300, + /*row_range_start=*/0, /*row_range_end=*/299, corrupt_source_meta); PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles(7, "btree", data_files, {payload}); ASSERT_TRUE(state.Groups().empty()); diff --git a/src/paimon/core/index/pksorted/pk_sorted_index_group.h b/src/paimon/core/index/pksorted/pk_sorted_index_group.h index 9680735dc..56d085067 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_index_group.h +++ b/src/paimon/core/index/pksorted/pk_sorted_index_group.h @@ -30,16 +30,16 @@ #include "paimon/core/index/pk/primary_key_index_source_meta.h" namespace paimon { -/// The single validated payload group which indexes one complete data level. +/// One validated payload group which indexes an immutable source subset at one data level. /// -/// A group is only created when the payload provably covers the current active source set -/// of its data level: exactly one payload, unique source names, source order / file names / -/// row counts identical to the expected level sources, matching index type and field id, -/// a row range of exactly `[0, total source rows - 1]` and a payload row count equal to the -/// source row count sum. Anything else must be treated as uncovered. +/// A group is only created when the payload provably covers its declared source set: unique +/// source names, matching index type and field id, a row range of exactly +/// `[0, total source rows - 1]`, and a payload row count equal to the source row count sum. +/// Snapshot-level active-source and overlap validation is performed by +/// `PkSortedBucketIndexState` before creating the group. class PkSortedIndexGroup { public: - /// Validates one payload against the expected level sources; returns null when any + /// Validates one payload against its expected source group; returns null when any /// coverage condition fails. static std::shared_ptr Create( int32_t field_id, const std::string& index_type, diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp index 07796b3f0..f89111bed 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp @@ -34,6 +34,7 @@ #include "paimon/common/utils/scope_guard.h" #include "paimon/core/global_index/global_index_evaluator_impl.h" #include "paimon/core/index/pk/primary_key_index_definitions.h" +#include "paimon/core/index/pk/primary_key_index_source_policy.h" #include "paimon/core/index/pksorted/pk_sorted_bucket_index_state.h" #include "paimon/core/manifest/file_kind.h" #include "paimon/global_index/bitmap_global_index_result.h" @@ -387,7 +388,9 @@ Result PrimaryKeySortedIndexScan::CreatePlan( } std::set> active_source_files; for (const std::shared_ptr& data_file : bucket_entry.second) { - active_source_files.emplace(data_file->file_name, data_file->row_count); + if (data_file != nullptr && PrimaryKeyIndexSourcePolicy::ShouldRead(*data_file)) { + active_source_files.emplace(data_file->file_name, data_file->row_count); + } } std::map>> groups_by_source; diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp index 4126232a0..bbd456c2e 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp @@ -257,6 +257,23 @@ class PrimaryKeySortedIndexScanTest : public ::testing::Test { return BuildPayload(std::move(ordinals)); } + Result> MakeMetadataPayload( + const std::string& name, const std::vector& sources, + int32_t data_level = 5) { + int64_t row_count = 0; + for (const PrimaryKeyIndexSourceFile& source : sources) { + row_count += source.row_count; + } + PAIMON_ASSIGN_OR_RAISE(PrimaryKeyIndexSourceMeta source_meta, + PrimaryKeyIndexSourceMeta::Create(data_level, sources)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr source_meta_bytes, + source_meta.Serialize(pool_)); + return std::make_shared( + "btree", name, /*file_size=*/1, row_count, std::nullopt, std::nullopt, + GlobalIndexMeta(/*row_range_start=*/0, /*row_range_end=*/row_count - 1, kPriceFieldId, + std::nullopt, /*index_meta=*/nullptr, source_meta_bytes)); + } + std::shared_ptr MakeSplit( const std::vector>& files, bool raw_convertible, const std::vector>& deletion_files = {}) { @@ -436,6 +453,206 @@ TEST_F(PrimaryKeySortedIndexScanTest, GroupAndQueryAreSharedAcrossSourceFiles) { ASSERT_EQ(1, *equal_call_count); } +TEST_F(PrimaryKeySortedIndexScanTest, DisjointSameLevelGroupsUseIndependentZeroBasedOrdinals) { + const std::vector left_sources = {{"a0.parquet", 3}, + {"a1.parquet", 4}}; + const std::vector right_sources = {{"b0.parquet", 5}, + {"b1.parquet", 6}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr left_payload, + MakeMetadataPayload("left.index", left_sources)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr right_payload, + MakeMetadataPayload("right.index", right_sources)); + + std::shared_ptr split = + MakeSplit({MakeDataFile("a0.parquet", 3, 5, FileSource::Compact()), + MakeDataFile("a1.parquet", 4, 5, FileSource::Compact()), + MakeDataFile("b0.parquet", 5, 5, FileSource::Compact()), + MakeDataFile("b1.parquet", 6, 5, FileSource::Compact())}, + /*raw_convertible=*/true); + std::vector entries = { + IndexManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0, left_payload), + IndexManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0, right_payload)}; + ASSERT_OK_AND_ASSIGN( + PrimaryKeySortedIndexScan::Plan plan, + PrimaryKeySortedIndexScan::CreatePlan(kSnapshotId, {split}, definitions_, entries)); + ASSERT_EQ(4, plan.Files().size()); + ASSERT_EQ(plan.Files()[0].Group(kPriceFieldId), plan.Files()[1].Group(kPriceFieldId)); + ASSERT_EQ(plan.Files()[2].Group(kPriceFieldId), plan.Files()[3].Group(kPriceFieldId)); + ASSERT_NE(plan.Files()[0].Group(kPriceFieldId), plan.Files()[2].Group(kPriceFieldId)); + + RoaringBitmap64 group_ordinal_zero; + group_ordinal_zero.Add(0); + auto factory_calls = std::make_shared>(); + PrimaryKeySortedIndexScan::ReaderFactory reader_factory = + [group_ordinal_zero, factory_calls]( + const PrimaryKeySortedIndexScan::FilePlan& file, + const PrimaryKeyIndexDefinition& definition, + const PkSortedIndexGroup& group) -> Result> { + (*factory_calls)[group.Payload()->FileName()]++; + return std::make_shared(group_ordinal_zero); + }; + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::EvaluatedPlan evaluated, + PrimaryKeySortedIndexScan::Evaluate(plan, table_schema_, PriceEqual(10), + definitions_, reader_factory)); + ASSERT_EQ(1, factory_calls->at("left.index")); + ASSERT_EQ(1, factory_calls->at("right.index")); + ASSERT_OK_AND_ASSIGN(std::vector> result_splits, + PrimaryKeySortedIndexResult::ToSplits(evaluated)); + ASSERT_EQ(2, result_splits.size()); + for (size_t i = 0; i < result_splits.size(); i++) { + auto indexed = std::dynamic_pointer_cast(result_splits[i]); + ASSERT_TRUE(indexed != nullptr); + auto inner = std::dynamic_pointer_cast(indexed->GetDataSplit()); + ASSERT_TRUE(inner != nullptr); + ASSERT_EQ(1, inner->DataFiles().size()); + ASSERT_EQ(i == 0 ? "a0.parquet" : "b0.parquet", inner->DataFiles()[0]->file_name); + ASSERT_EQ(1, indexed->RowRanges().size()); + ASSERT_EQ(0, indexed->RowRanges()[0].from); + ASSERT_EQ(0, indexed->RowRanges()[0].to); + } +} + +TEST_F(PrimaryKeySortedIndexScanTest, RetiredSourcesPreserveGroupOrdinalOffsets) { + const std::vector sources = {{"a-retired-prefix.parquet", 3}, + {"b-active-left.parquet", 4}, + {"c-retired-middle.parquet", 5}, + {"d-active-right.parquet", 6}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr payload, + MakeMetadataPayload("payload.index", sources)); + std::shared_ptr split = + MakeSplit({MakeDataFile("b-active-left.parquet", 4, 5, FileSource::Compact()), + MakeDataFile("d-active-right.parquet", 6, 5, FileSource::Compact())}, + /*raw_convertible=*/true); + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::Plan plan, + PrimaryKeySortedIndexScan::CreatePlan(kSnapshotId, {split}, definitions_, + MakeEntries(payload))); + ASSERT_EQ(2, plan.Files().size()); + ASSERT_EQ(plan.Files()[0].Group(kPriceFieldId), plan.Files()[1].Group(kPriceFieldId)); + ASSERT_EQ(sources, plan.Files()[0].Group(kPriceFieldId)->SourceFiles()); + + RoaringBitmap64 group_positions; + group_positions.Add(3); + group_positions.Add(12); + auto equal_call_count = std::make_shared(0); + PrimaryKeySortedIndexScan::ReaderFactory reader_factory = + [group_positions, equal_call_count]( + const PrimaryKeySortedIndexScan::FilePlan& file, + const PrimaryKeyIndexDefinition& definition, + const PkSortedIndexGroup& group) -> Result> { + return std::make_shared(group_positions, equal_call_count); + }; + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::EvaluatedPlan evaluated, + PrimaryKeySortedIndexScan::Evaluate(plan, table_schema_, PriceEqual(10), + definitions_, reader_factory)); + ASSERT_EQ(1, *equal_call_count); + ASSERT_OK_AND_ASSIGN(std::vector> result_splits, + PrimaryKeySortedIndexResult::ToSplits(evaluated)); + ASSERT_EQ(2, result_splits.size()); + for (size_t i = 0; i < result_splits.size(); i++) { + auto indexed = std::dynamic_pointer_cast(result_splits[i]); + ASSERT_TRUE(indexed != nullptr); + auto inner = std::dynamic_pointer_cast(indexed->GetDataSplit()); + ASSERT_TRUE(inner != nullptr); + ASSERT_EQ(1, inner->DataFiles().size()); + ASSERT_EQ(i == 0 ? "b-active-left.parquet" : "d-active-right.parquet", + inner->DataFiles()[0]->file_name); + ASSERT_EQ(1, indexed->RowRanges().size()); + ASSERT_EQ(0, indexed->RowRanges()[0].from); + ASSERT_EQ(0, indexed->RowRanges()[0].to); + } +} + +TEST_F(PrimaryKeySortedIndexScanTest, UpdatedFileFallsBackWhileOldIndexedFileKeepsDeletionFile) { + const std::vector old_sources = {{"a-retired-prefix.parquet", 3}, + {"b-active-old.parquet", 4}, + {"c-retired-middle.parquet", 5}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr payload, + MakeMetadataPayload("old-payload.index", old_sources)); + DeletionFile old_deletion_file("dv-old", /*offset=*/0, /*length=*/16, /*cardinality=*/1); + std::shared_ptr split = MakeSplit( + {MakeDataFile("b-active-old.parquet", 4, 5, FileSource::Compact()), + MakeDataFile("d-new.parquet", 6, 5, FileSource::Compact())}, + /*raw_convertible=*/true, {std::optional(old_deletion_file), std::nullopt}); + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::Plan plan, + PrimaryKeySortedIndexScan::CreatePlan(kSnapshotId, {split}, definitions_, + MakeEntries(payload))); + ASSERT_EQ(2, plan.Files().size()); + ASSERT_TRUE(plan.Files()[0].Group(kPriceFieldId) != nullptr); + ASSERT_TRUE(plan.Files()[1].Groups().empty()); + + RoaringBitmap64 old_group_position; + old_group_position.Add(4); + PrimaryKeySortedIndexScan::ReaderFactory reader_factory = + [old_group_position]( + const PrimaryKeySortedIndexScan::FilePlan& file, + const PrimaryKeyIndexDefinition& definition, + const PkSortedIndexGroup& group) -> Result> { + return std::make_shared(old_group_position); + }; + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::EvaluatedPlan evaluated, + PrimaryKeySortedIndexScan::Evaluate(plan, table_schema_, PriceEqual(10), + definitions_, reader_factory)); + ASSERT_OK_AND_ASSIGN(std::vector> result_splits, + PrimaryKeySortedIndexResult::ToSplits(evaluated)); + ASSERT_EQ(2, result_splits.size()); + + auto indexed_old = std::dynamic_pointer_cast(result_splits[0]); + ASSERT_TRUE(indexed_old != nullptr); + ASSERT_EQ(1, indexed_old->RowRanges().size()); + ASSERT_EQ(1, indexed_old->RowRanges()[0].from); + ASSERT_EQ(1, indexed_old->RowRanges()[0].to); + auto old_file = std::dynamic_pointer_cast(indexed_old->GetDataSplit()); + ASSERT_TRUE(old_file != nullptr); + ASSERT_EQ("b-active-old.parquet", old_file->DataFiles()[0]->file_name); + ASSERT_EQ(1, old_file->DeletionFiles().size()); + ASSERT_TRUE(old_file->DeletionFiles()[0].has_value()); + ASSERT_EQ("dv-old", old_file->DeletionFiles()[0]->path); + + auto new_file = std::dynamic_pointer_cast(result_splits[1]); + ASSERT_TRUE(new_file != nullptr); + ASSERT_EQ("d-new.parquet", new_file->DataFiles()[0]->file_name); + ASSERT_FALSE(new_file->RawConvertible()); + ASSERT_EQ(1, new_file->DeletionFiles().size()); + ASSERT_FALSE(new_file->DeletionFiles()[0].has_value()); +} + +TEST_F(PrimaryKeySortedIndexScanTest, NoneligibleActiveSourceDoesNotInheritAcceptedGroup) { + const std::vector sources = {{"a-eligible.parquet", 4}, + {"b-append.parquet", 6}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr payload, + MakeMetadataPayload("payload.index", sources)); + std::shared_ptr split = + MakeSplit({MakeDataFile("a-eligible.parquet", 4, 5, FileSource::Compact()), + MakeDataFile("b-append.parquet", 6, 5, FileSource::Append())}, + /*raw_convertible=*/true); + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::Plan plan, + PrimaryKeySortedIndexScan::CreatePlan(kSnapshotId, {split}, definitions_, + MakeEntries(payload))); + ASSERT_EQ(2, plan.Files().size()); + ASSERT_TRUE(plan.Files()[0].Group(kPriceFieldId) != nullptr); + ASSERT_TRUE(plan.Files()[1].Groups().empty()); + + RoaringBitmap64 eligible_position; + eligible_position.Add(0); + PrimaryKeySortedIndexScan::ReaderFactory reader_factory = + [eligible_position]( + const PrimaryKeySortedIndexScan::FilePlan& file, + const PrimaryKeyIndexDefinition& definition, + const PkSortedIndexGroup& group) -> Result> { + return std::make_shared(eligible_position); + }; + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::EvaluatedPlan evaluated, + PrimaryKeySortedIndexScan::Evaluate(plan, table_schema_, PriceEqual(10), + definitions_, reader_factory)); + ASSERT_OK_AND_ASSIGN(std::vector> result_splits, + PrimaryKeySortedIndexResult::ToSplits(evaluated)); + ASSERT_EQ(2, result_splits.size()); + ASSERT_TRUE(std::dynamic_pointer_cast(result_splits[0]) != nullptr); + auto append_fallback = std::dynamic_pointer_cast(result_splits[1]); + ASSERT_TRUE(append_fallback != nullptr); + ASSERT_EQ("b-append.parquet", append_fallback->DataFiles()[0]->file_name); +} + TEST_F(PrimaryKeySortedIndexScanTest, EmptyResultOmitsAllFiles) { ASSERT_OK_AND_ASSIGN(std::shared_ptr payload, BuildPayload()); std::shared_ptr split = From 9a1f257dfe757d85bb7ce23bb549983b7c01d039 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Fri, 28 Aug 2026 21:27:58 -0400 Subject: [PATCH 2/3] fix(index): retain one PK index group per level --- .../user_guide/primary_key_global_index.rst | 10 +-- .../pksorted/pk_sorted_bucket_index_state.cpp | 82 +++++++------------ .../pksorted/pk_sorted_bucket_index_state.h | 12 +-- .../pk_sorted_bucket_index_state_test.cpp | 70 +++++----------- .../index/pksorted/pk_sorted_index_group.h | 4 +- .../primary_key_sorted_index_scan_test.cpp | 59 ------------- 6 files changed, 63 insertions(+), 174 deletions(-) diff --git a/docs/source/user_guide/primary_key_global_index.rst b/docs/source/user_guide/primary_key_global_index.rst index 8964ee5a1..dcba88fb9 100644 --- a/docs/source/user_guide/primary_key_global_index.rst +++ b/docs/source/user_guide/primary_key_global_index.rst @@ -45,11 +45,11 @@ Semantics - A payload retains its complete ordered source list as the group-ordinal namespace. If part of that list is not in the current scan because it was retired or safely pruned, - the payload can still cover the remaining files. Several payloads may coexist when their - active source intersections are disjoint. A payload is rejected if it has no active - source, an active source's row count differs, its metadata or row range is invalid, or - another payload claims the same active source. Active files without accepted coverage - are scanned normally. + the payload can still cover the remaining files. There is at most one accepted payload + per data level. A payload is rejected if it has no active source at its metadata-declared + level, an active source's row count differs, its metadata or row range is invalid, or + another payload exists for that level. Active files without accepted coverage are scanned + normally. - ``AND`` predicates narrow with any safely evaluable indexed child; ``OR`` predicates only use the index when every branch is evaluable. Files whose evaluation fails, whose positions are out of range, or whose result needs more than 4096 ranges fall back to a diff --git a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp index ca2e5dc9a..3d67432cc 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp +++ b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp @@ -36,7 +36,6 @@ struct PayloadCandidate { std::shared_ptr payload; std::shared_ptr group; std::vector active_sources; - bool conflicted = false; }; } // namespace @@ -59,20 +58,9 @@ PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles( }); } - std::map active_sources; - std::set ambiguous_active_source_names; - for (const auto& level_sources : sources_by_level) { - for (const PrimaryKeyIndexSourceFile& source : level_sources.second) { - if (!active_sources.emplace(source.file_name, source.row_count).second) { - ambiguous_active_source_names.insert(source.file_name); - } - } - } - // Keep the complete immutable source group for ordinal localization, but only claim the - // sources which are still active in this snapshot. Several disjoint groups may coexist - // after an update; no active source file may be claimed by two groups. - std::vector candidates; + // sources which are still active at the metadata-declared level in this snapshot. + std::map> candidates_by_level; std::vector> rejected; for (const std::shared_ptr& payload : active_payloads) { if (payload == nullptr) { @@ -91,21 +79,30 @@ PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles( continue; } PrimaryKeyIndexSourceMeta source_meta = std::move(source_meta_result).value(); + auto active_level = sources_by_level.find(source_meta.DataLevel()); + if (active_level == sources_by_level.end()) { + rejected.push_back(payload); + continue; + } const std::vector& payload_sources = source_meta.SourceFiles(); bool valid_candidate = !payload_sources.empty(); std::vector active_intersection; + size_t active_source_index = 0; for (size_t i = 0; valid_candidate && i < payload_sources.size(); i++) { const PrimaryKeyIndexSourceFile& source = payload_sources[i]; if (i > 0 && payload_sources[i - 1].file_name >= source.file_name) { valid_candidate = false; break; } - auto active_source = active_sources.find(source.file_name); - if (active_source == active_sources.end()) { + while (active_source_index < active_level->second.size() && + active_level->second[active_source_index].file_name < source.file_name) { + active_source_index++; + } + if (active_source_index == active_level->second.size() || + active_level->second[active_source_index].file_name != source.file_name) { continue; } - if (active_source->second != source.row_count || - ambiguous_active_source_names.count(source.file_name) > 0) { + if (active_level->second[active_source_index].row_count != source.row_count) { valid_candidate = false; break; } @@ -121,54 +118,35 @@ PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles( rejected.push_back(payload); continue; } - candidates.push_back( - {payload, std::move(group), std::move(active_intersection), /*conflicted=*/false}); - } - - std::map, std::vector> candidates_by_source; - for (size_t i = 0; i < candidates.size(); i++) { - const PayloadCandidate& candidate = candidates[i]; - for (const PrimaryKeyIndexSourceFile& source : candidate.active_sources) { - candidates_by_source[{source.file_name, source.row_count}].push_back(i); - } - } - for (const auto& source_candidates : candidates_by_source) { - if (source_candidates.second.size() > 1) { - for (size_t candidate_index : source_candidates.second) { - candidates[candidate_index].conflicted = true; - } - } + candidates_by_level[source_meta.DataLevel()].push_back( + {payload, std::move(group), std::move(active_intersection)}); } std::vector> groups; - std::set> covered_sources; - for (PayloadCandidate& candidate : candidates) { - if (candidate.conflicted) { - rejected.push_back(std::move(candidate.payload)); + std::map>> covered_sources_by_level; + for (auto& level_candidates : candidates_by_level) { + if (level_candidates.second.size() != 1) { + for (PayloadCandidate& candidate : level_candidates.second) { + rejected.push_back(std::move(candidate.payload)); + } continue; } + PayloadCandidate& candidate = level_candidates.second[0]; for (const PrimaryKeyIndexSourceFile& source : candidate.active_sources) { - covered_sources.emplace(source.file_name, source.row_count); + covered_sources_by_level[level_candidates.first].emplace(source.file_name, + source.row_count); } groups.push_back(std::move(candidate.group)); } - std::sort(groups.begin(), groups.end(), - [](const std::shared_ptr& left, - const std::shared_ptr& right) { - if (left->DataLevel() != right->DataLevel()) { - return left->DataLevel() < right->DataLevel(); - } - return left->SourceFiles().front().file_name < - right->SourceFiles().front().file_name; - }); std::vector covered; std::vector uncovered; for (const auto& level_sources : sources_by_level) { for (const PrimaryKeyIndexSourceFile& source : level_sources.second) { - auto& target = covered_sources.count({source.file_name, source.row_count}) > 0 - ? covered - : uncovered; + auto covered_level = covered_sources_by_level.find(level_sources.first); + bool is_covered = covered_level != covered_sources_by_level.end() && + covered_level->second.count({source.file_name, source.row_count}) > 0; + auto& target = is_covered ? covered : uncovered; target.push_back(source); } } diff --git a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h index 89474d015..8dd558050 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h +++ b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.h @@ -34,12 +34,12 @@ namespace paimon { /// Immutable source-backed sorted-index state for one field and bucket. /// /// Derives the eligible source sets from the active data files and keeps validated, -/// non-overlapping payload groups. A group retains its complete immutable source list for -/// ordinal localization while covering only the listed files which are still active. A -/// payload is rejected when its metadata is invalid, it has no active source, an active -/// source has a different row count, its source order is not canonical, or it overlaps -/// another candidate on an active source. Active files without an accepted group remain -/// uncovered and must be scanned normally. +/// payload groups. A group retains its complete immutable source list for ordinal +/// localization while covering only the listed files which are still active at its +/// metadata-declared level. A payload is rejected when its metadata is invalid, it has no +/// active source at that level, an active source has a different row count, its source +/// order is not canonical, or another payload exists for the same level. Active files +/// without an accepted group remain uncovered and must be scanned normally. class PkSortedBucketIndexState { public: static PkSortedBucketIndexState FromActiveDataFiles( diff --git a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp index 06f93dbe0..55bfc4896 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp +++ b/src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp @@ -190,7 +190,7 @@ TEST_F(PkSortedBucketIndexStateTest, AcceptsActiveSubsetAndLeavesOtherFilesUncov ASSERT_TRUE(state.RejectedPayloads().empty()); } -TEST_F(PkSortedBucketIndexStateTest, AcceptsDisjointPayloadGroupsAtSameLevel) { +TEST_F(PkSortedBucketIndexStateTest, RejectsMultiplePayloadGroupsAtSameLevel) { std::vector> data_files = { MakeDataFile("a", 100, 5, FileSource::Compact()), MakeDataFile("b", 200, 5, FileSource::Compact()), @@ -201,13 +201,11 @@ TEST_F(PkSortedBucketIndexStateTest, AcceptsDisjointPayloadGroupsAtSameLevel) { MakeNamedPayload("second.index", 7, "btree", 5, {{"c", 50}}, 50); PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( 7, "btree", data_files, {second_payload, first_payload}); - ASSERT_EQ(2, state.Groups().size()); - ASSERT_EQ(first_payload, state.Groups()[0]->Payload()); - ASSERT_EQ(second_payload, state.Groups()[1]->Payload()); + ASSERT_TRUE(state.Groups().empty()); + ASSERT_TRUE(state.CoveredSourceFiles().empty()); ASSERT_EQ((std::vector{{"a", 100}, {"b", 200}, {"c", 50}}), - state.CoveredSourceFiles()); - ASSERT_TRUE(state.UncoveredSourceFiles().empty()); - ASSERT_TRUE(state.RejectedPayloads().empty()); + state.UncoveredSourceFiles()); + ASSERT_EQ(2, state.RejectedPayloads().size()); } TEST_F(PkSortedBucketIndexStateTest, RetainsRetiredSourcesButCoversOnlyActiveIntersection) { @@ -216,18 +214,14 @@ TEST_F(PkSortedBucketIndexStateTest, RetainsRetiredSourcesButCoversOnlyActiveInt MakeDataFile("b", 200, 5, FileSource::Compact())}; std::shared_ptr mixed_payload = MakeNamedPayload("mixed.index", 7, "btree", 5, {{"a", 100}, {"retired", 50}}, 150); - std::shared_ptr valid_payload = - MakeNamedPayload("valid.index", 7, "btree", 5, {{"b", 200}}, 200); - PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( - 7, "btree", data_files, {mixed_payload, valid_payload}); - ASSERT_EQ(2, state.Groups().size()); + PkSortedBucketIndexState state = + PkSortedBucketIndexState::FromActiveDataFiles(7, "btree", data_files, {mixed_payload}); + ASSERT_EQ(1, state.Groups().size()); ASSERT_EQ(mixed_payload, state.Groups()[0]->Payload()); ASSERT_EQ((std::vector{{"a", 100}, {"retired", 50}}), state.Groups()[0]->SourceFiles()); - ASSERT_EQ(valid_payload, state.Groups()[1]->Payload()); - ASSERT_EQ((std::vector{{"a", 100}, {"b", 200}}), - state.CoveredSourceFiles()); - ASSERT_TRUE(state.UncoveredSourceFiles().empty()); + ASSERT_EQ((std::vector{{"a", 100}}), state.CoveredSourceFiles()); + ASSERT_EQ((std::vector{{"b", 200}}), state.UncoveredSourceFiles()); ASSERT_TRUE(state.RejectedPayloads().empty()); } @@ -246,43 +240,19 @@ TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWhoseSourcesAreAllRetired) { state.RejectedPayloads()); } -TEST_F(PkSortedBucketIndexStateTest, RetiredSourceOverlapDoesNotConflict) { +TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadForDifferentActiveDataLevel) { std::vector> data_files = { - MakeDataFile("b-active", 100, 5, FileSource::Compact()), - MakeDataFile("c-active", 200, 5, FileSource::Compact())}; - std::shared_ptr left_payload = - MakeNamedPayload("left.index", 7, "btree", 5, {{"a-retired", 50}, {"b-active", 100}}, 150); - std::shared_ptr right_payload = - MakeNamedPayload("right.index", 7, "btree", 5, {{"a-retired", 50}, {"c-active", 200}}, 250); + MakeDataFile("active", 100, 4, FileSource::Compact())}; + std::shared_ptr wrong_level_payload = + MakePayload(7, "btree", 5, {{"active", 100}}, 100); PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( - 7, "btree", data_files, {left_payload, right_payload}); - ASSERT_EQ(2, state.Groups().size()); - ASSERT_EQ((std::vector{{"b-active", 100}, {"c-active", 200}}), - state.CoveredSourceFiles()); - ASSERT_TRUE(state.UncoveredSourceFiles().empty()); - ASSERT_TRUE(state.RejectedPayloads().empty()); -} - -TEST_F(PkSortedBucketIndexStateTest, RejectsOverlappingGroupsButKeepsDisjointGroup) { - std::vector> data_files = { - MakeDataFile("a", 100, 5, FileSource::Compact()), - MakeDataFile("b", 200, 5, FileSource::Compact()), - MakeDataFile("c", 50, 5, FileSource::Compact()), - MakeDataFile("d", 75, 5, FileSource::Compact())}; - std::shared_ptr left_payload = - MakeNamedPayload("left.index", 7, "btree", 5, {{"a", 100}, {"b", 200}}, 300); - std::shared_ptr right_payload = - MakeNamedPayload("right.index", 7, "btree", 5, {{"b", 200}, {"c", 50}}, 250); - std::shared_ptr disjoint_payload = - MakeNamedPayload("disjoint.index", 7, "btree", 5, {{"d", 75}}, 75); - PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles( - 7, "btree", data_files, {left_payload, disjoint_payload, right_payload}); - ASSERT_EQ(1, state.Groups().size()); - ASSERT_EQ(disjoint_payload, state.Groups()[0]->Payload()); - ASSERT_EQ((std::vector{{"d", 75}}), state.CoveredSourceFiles()); - ASSERT_EQ((std::vector{{"a", 100}, {"b", 200}, {"c", 50}}), + 7, "btree", data_files, {wrong_level_payload}); + ASSERT_TRUE(state.Groups().empty()); + ASSERT_TRUE(state.CoveredSourceFiles().empty()); + ASSERT_EQ((std::vector{{"active", 100}}), state.UncoveredSourceFiles()); - ASSERT_EQ(2, state.RejectedPayloads().size()); + ASSERT_EQ((std::vector>{wrong_level_payload}), + state.RejectedPayloads()); } TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadsClaimingTheSameActiveSources) { diff --git a/src/paimon/core/index/pksorted/pk_sorted_index_group.h b/src/paimon/core/index/pksorted/pk_sorted_index_group.h index 56d085067..bf8fb5649 100644 --- a/src/paimon/core/index/pksorted/pk_sorted_index_group.h +++ b/src/paimon/core/index/pksorted/pk_sorted_index_group.h @@ -30,12 +30,12 @@ #include "paimon/core/index/pk/primary_key_index_source_meta.h" namespace paimon { -/// One validated payload group which indexes an immutable source subset at one data level. +/// The single validated payload group which indexes an immutable source set at one data level. /// /// A group is only created when the payload provably covers its declared source set: unique /// source names, matching index type and field id, a row range of exactly /// `[0, total source rows - 1]`, and a payload row count equal to the source row count sum. -/// Snapshot-level active-source and overlap validation is performed by +/// Snapshot-level active-source and one-group-per-level validation is performed by /// `PkSortedBucketIndexState` before creating the group. class PkSortedIndexGroup { public: diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp index bbd456c2e..2c6e5d876 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp @@ -453,65 +453,6 @@ TEST_F(PrimaryKeySortedIndexScanTest, GroupAndQueryAreSharedAcrossSourceFiles) { ASSERT_EQ(1, *equal_call_count); } -TEST_F(PrimaryKeySortedIndexScanTest, DisjointSameLevelGroupsUseIndependentZeroBasedOrdinals) { - const std::vector left_sources = {{"a0.parquet", 3}, - {"a1.parquet", 4}}; - const std::vector right_sources = {{"b0.parquet", 5}, - {"b1.parquet", 6}}; - ASSERT_OK_AND_ASSIGN(std::shared_ptr left_payload, - MakeMetadataPayload("left.index", left_sources)); - ASSERT_OK_AND_ASSIGN(std::shared_ptr right_payload, - MakeMetadataPayload("right.index", right_sources)); - - std::shared_ptr split = - MakeSplit({MakeDataFile("a0.parquet", 3, 5, FileSource::Compact()), - MakeDataFile("a1.parquet", 4, 5, FileSource::Compact()), - MakeDataFile("b0.parquet", 5, 5, FileSource::Compact()), - MakeDataFile("b1.parquet", 6, 5, FileSource::Compact())}, - /*raw_convertible=*/true); - std::vector entries = { - IndexManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0, left_payload), - IndexManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), /*bucket=*/0, right_payload)}; - ASSERT_OK_AND_ASSIGN( - PrimaryKeySortedIndexScan::Plan plan, - PrimaryKeySortedIndexScan::CreatePlan(kSnapshotId, {split}, definitions_, entries)); - ASSERT_EQ(4, plan.Files().size()); - ASSERT_EQ(plan.Files()[0].Group(kPriceFieldId), plan.Files()[1].Group(kPriceFieldId)); - ASSERT_EQ(plan.Files()[2].Group(kPriceFieldId), plan.Files()[3].Group(kPriceFieldId)); - ASSERT_NE(plan.Files()[0].Group(kPriceFieldId), plan.Files()[2].Group(kPriceFieldId)); - - RoaringBitmap64 group_ordinal_zero; - group_ordinal_zero.Add(0); - auto factory_calls = std::make_shared>(); - PrimaryKeySortedIndexScan::ReaderFactory reader_factory = - [group_ordinal_zero, factory_calls]( - const PrimaryKeySortedIndexScan::FilePlan& file, - const PrimaryKeyIndexDefinition& definition, - const PkSortedIndexGroup& group) -> Result> { - (*factory_calls)[group.Payload()->FileName()]++; - return std::make_shared(group_ordinal_zero); - }; - ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::EvaluatedPlan evaluated, - PrimaryKeySortedIndexScan::Evaluate(plan, table_schema_, PriceEqual(10), - definitions_, reader_factory)); - ASSERT_EQ(1, factory_calls->at("left.index")); - ASSERT_EQ(1, factory_calls->at("right.index")); - ASSERT_OK_AND_ASSIGN(std::vector> result_splits, - PrimaryKeySortedIndexResult::ToSplits(evaluated)); - ASSERT_EQ(2, result_splits.size()); - for (size_t i = 0; i < result_splits.size(); i++) { - auto indexed = std::dynamic_pointer_cast(result_splits[i]); - ASSERT_TRUE(indexed != nullptr); - auto inner = std::dynamic_pointer_cast(indexed->GetDataSplit()); - ASSERT_TRUE(inner != nullptr); - ASSERT_EQ(1, inner->DataFiles().size()); - ASSERT_EQ(i == 0 ? "a0.parquet" : "b0.parquet", inner->DataFiles()[0]->file_name); - ASSERT_EQ(1, indexed->RowRanges().size()); - ASSERT_EQ(0, indexed->RowRanges()[0].from); - ASSERT_EQ(0, indexed->RowRanges()[0].to); - } -} - TEST_F(PrimaryKeySortedIndexScanTest, RetiredSourcesPreserveGroupOrdinalOffsets) { const std::vector sources = {{"a-retired-prefix.parquet", 3}, {"b-active-left.parquet", 4}, From 7059536b89702feb60cf038755e25de50ee65328 Mon Sep 17 00:00:00 2001 From: "wangyong.alen" Date: Fri, 28 Aug 2026 23:55:03 -0400 Subject: [PATCH 3/3] fix(index): bind source groups to their data level --- .../source/primary_key_sorted_index_scan.cpp | 10 ++++++---- .../primary_key_sorted_index_scan_test.cpp | 19 +++++++++++++++++++ 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp index f89111bed..a523b920d 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan.cpp @@ -22,6 +22,7 @@ #include #include #include +#include #include #include @@ -386,10 +387,11 @@ Result PrimaryKeySortedIndexScan::CreatePlan( if (payloads_iter != payloads_by_bucket.end()) { bucket_payloads = payloads_iter->second; } - std::set> active_source_files; + std::set> active_source_files; for (const std::shared_ptr& data_file : bucket_entry.second) { if (data_file != nullptr && PrimaryKeyIndexSourcePolicy::ShouldRead(*data_file)) { - active_source_files.emplace(data_file->file_name, data_file->row_count); + active_source_files.emplace(data_file->level, data_file->file_name, + data_file->row_count); } } std::map>> @@ -408,8 +410,8 @@ Result PrimaryKeySortedIndexScan::CreatePlan( definition_payloads); for (const std::shared_ptr& group : state.Groups()) { for (const PrimaryKeyIndexSourceFile& source_file : group->SourceFiles()) { - if (active_source_files.count({source_file.file_name, source_file.row_count}) == - 0) { + if (active_source_files.count({group->DataLevel(), source_file.file_name, + source_file.row_count}) == 0) { continue; } groups_by_source[source_file.file_name][definition.FieldId()] = group; diff --git a/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp b/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp index 2c6e5d876..7b14a5c4f 100644 --- a/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp +++ b/src/paimon/core/table/source/primary_key_sorted_index_scan_test.cpp @@ -503,6 +503,25 @@ TEST_F(PrimaryKeySortedIndexScanTest, RetiredSourcesPreserveGroupOrdinalOffsets) } } +TEST_F(PrimaryKeySortedIndexScanTest, SourceMovedToDifferentLevelFallsBack) { + const std::vector sources = {{"a-stays.parquet", 4}, + {"b-moved.parquet", 6}}; + ASSERT_OK_AND_ASSIGN(std::shared_ptr payload, + MakeMetadataPayload("payload.index", sources, /*data_level=*/5)); + std::shared_ptr split = + MakeSplit({MakeDataFile("a-stays.parquet", 4, 5, FileSource::Compact()), + MakeDataFile("b-moved.parquet", 6, 4, FileSource::Compact())}, + /*raw_convertible=*/true); + + ASSERT_OK_AND_ASSIGN(PrimaryKeySortedIndexScan::Plan plan, + PrimaryKeySortedIndexScan::CreatePlan(kSnapshotId, {split}, definitions_, + MakeEntries(payload))); + + ASSERT_EQ(2, plan.Files().size()); + ASSERT_NE(nullptr, plan.Files()[0].Group(kPriceFieldId)); + ASSERT_EQ(nullptr, plan.Files()[1].Group(kPriceFieldId)); +} + TEST_F(PrimaryKeySortedIndexScanTest, UpdatedFileFallsBackWhileOldIndexedFileKeepsDeletionFile) { const std::vector old_sources = {{"a-retired-prefix.parquet", 3}, {"b-active-old.parquet", 4},