perf(index): reuse primary-key index groups with retired source files

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. 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

src/paimon/core/index/pksorted/pk_sorted_bucket_index_state.cpp

@@ -23,13 +23,22 @@
 #include <map>
 #include <optional>
 #include <set>
+#include <string>
 #include <utility>
 
 #include "paimon/core/index/global_index_meta.h"
 #include "paimon/core/index/pk/primary_key_index_source_meta.h"
 #include "paimon/core/index/pk/primary_key_index_source_policy.h"
 
 namespace paimon {
+namespace {
+struct PayloadCandidate {
+    std::shared_ptr<IndexFileMeta> payload;
+    std::shared_ptr<PkSortedIndexGroup> group;
+    std::vector<PrimaryKeyIndexSourceFile> active_sources;
+};
+}  // namespace
+
 PkSortedBucketIndexState PkSortedBucketIndexState::FromActiveDataFiles(
     int32_t field_id, const std::string& index_type,
     const std::vector<std::shared_ptr<DataFileMeta>>& active_data_files,
@@ -49,10 +58,9 @@ 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<int32_t, std::vector<std::shared_ptr<IndexFileMeta>>> payloads_by_level;
-    std::map<int32_t, std::vector<PrimaryKeyIndexSourceMeta>> payload_metas_by_level;
+    // Keep the complete immutable source group for ordinal localization, but only claim the
+    // sources which are still active at the metadata-declared level in this snapshot.
+    std::map<int32_t, std::vector<PayloadCandidate>> candidates_by_level;
     std::vector<std::shared_ptr<IndexFileMeta>> rejected;
     for (const std::shared_ptr<IndexFileMeta>& payload : active_payloads) {
         if (payload == nullptr) {
@@ -71,39 +79,76 @@ 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()) {
+        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<PrimaryKeyIndexSourceFile>& payload_sources = source_meta.SourceFiles();
+        bool valid_candidate = !payload_sources.empty();
+        std::vector<PrimaryKeyIndexSourceFile> 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;
+            }
+            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_level->second[active_source_index].row_count != source.row_count) {
+                valid_candidate = false;
+                break;
+            }
+            active_intersection.push_back(source);
+        }
+        if (!valid_candidate || active_intersection.empty()) {
             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));
+        std::shared_ptr<PkSortedIndexGroup> group =
+            PkSortedIndexGroup::Create(field_id, index_type, payload_sources, payload, source_meta);
+        if (group == nullptr) {
+            rejected.push_back(payload);
+            continue;
+        }
+        candidates_by_level[source_meta.DataLevel()].push_back(
+            {payload, std::move(group), std::move(active_intersection)});
     }
 
     std::vector<std::shared_ptr<PkSortedIndexGroup>> groups;
-    std::set<int32_t> covered_levels;
-    for (const auto& level_payloads : payloads_by_level) {
-        int32_t level = level_payloads.first;
-        std::shared_ptr<PkSortedIndexGroup> 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::map<int32_t, std::set<std::pair<std::string, int64_t>>> 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;
         }
-        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());
+        PayloadCandidate& candidate = level_candidates.second[0];
+        for (const PrimaryKeyIndexSourceFile& source : candidate.active_sources) {
+            covered_sources_by_level[level_candidates.first].emplace(source.file_name,
+                                                                     source.row_count);
         }
+        groups.push_back(std::move(candidate.group));
     }
 
     std::vector<PrimaryKeyIndexSourceFile> covered;
     std::vector<PrimaryKeyIndexSourceFile> 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 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);
+        }
     }
     return PkSortedBucketIndexState(std::move(groups), std::move(covered), std::move(uncovered),
                                     std::move(rejected));

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,
+/// 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(

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<IndexFileMeta> MakePayload(
-        int32_t field_id, const std::string& index_type, int32_t data_level,
-        const std::vector<PrimaryKeyIndexSourceFile>& sources, int64_t total_row_count,
-        int64_t row_range_start, int64_t row_range_end) const {
+    std::shared_ptr<IndexFileMeta> MakeNamedPayload(
+        const std::string& payload_name, int32_t field_id, const std::string& index_type,
+        int32_t data_level, const std::vector<PrimaryKeyIndexSourceFile>& 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<Bytes> 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<IndexFileMeta> MakePayload(
+        int32_t field_id, const std::string& index_type, int32_t data_level,
+        const std::vector<PrimaryKeyIndexSourceFile>& 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<IndexFileMeta> MakeNamedPayload(
+        const std::string& payload_name, int32_t field_id, const std::string& index_type,
+        int32_t data_level, const std::vector<PrimaryKeyIndexSourceFile>& 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<IndexFileMeta> MakePayload(
         int32_t field_id, const std::string& index_type, int32_t data_level,
         const std::vector<PrimaryKeyIndexSourceFile>& sources, int64_t total_row_count) const {
@@ -77,13 +94,13 @@ class PkSortedBucketIndexStateTest : public ::testing::Test {
     }
 
     std::shared_ptr<IndexFileMeta> 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<Bytes>& 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<IndexFileMeta>(index_type, /*file_name=*/"payload.index",
+        return std::make_shared<IndexFileMeta>(index_type, payload_name,
                                                /*file_size=*/2048, total_row_count,
                                                /*dv_ranges=*/std::nullopt,
                                                /*external_path=*/std::nullopt, global_index_meta);
@@ -159,23 +176,86 @@ TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWithMismatchedSourceRowCount)
     ASSERT_EQ(2, state.UncoveredSourceFiles().size());
 }
 
-TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadsCoveringWrongSourceSet) {
+TEST_F(PkSortedBucketIndexStateTest, AcceptsActiveSubsetAndLeavesOtherFilesUncovered) {
     std::vector<std::shared_ptr<DataFileMeta>> data_files = {
         MakeDataFile("a", 100, 5, FileSource::Compact()),
         MakeDataFile("b", 200, 5, FileSource::Compact())};
-    std::shared_ptr<IndexFileMeta> missing_source_payload =
-        MakePayload(7, "btree", 5, {{"a", 100}}, 100);
-    std::shared_ptr<IndexFileMeta> extra_source_payload =
-        MakePayload(7, "btree", 5, {{"a", 100}, {"b", 200}, {"c", 50}}, 350);
+    std::shared_ptr<IndexFileMeta> 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<PrimaryKeyIndexSourceFile>{{"a", 100}}), state.CoveredSourceFiles());
+    ASSERT_EQ((std::vector<PrimaryKeyIndexSourceFile>{{"b", 200}}), state.UncoveredSourceFiles());
+    ASSERT_TRUE(state.RejectedPayloads().empty());
+}
+
+TEST_F(PkSortedBucketIndexStateTest, RejectsMultiplePayloadGroupsAtSameLevel) {
+    std::vector<std::shared_ptr<DataFileMeta>> data_files = {
+        MakeDataFile("a", 100, 5, FileSource::Compact()),
+        MakeDataFile("b", 200, 5, FileSource::Compact()),
+        MakeDataFile("c", 50, 5, FileSource::Compact())};
+    std::shared_ptr<IndexFileMeta> first_payload =
+        MakeNamedPayload("first.index", 7, "btree", 5, {{"a", 100}, {"b", 200}}, 300);
+    std::shared_ptr<IndexFileMeta> 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_TRUE(state.Groups().empty());
+    ASSERT_TRUE(state.CoveredSourceFiles().empty());
+    ASSERT_EQ((std::vector<PrimaryKeyIndexSourceFile>{{"a", 100}, {"b", 200}, {"c", 50}}),
+              state.UncoveredSourceFiles());
     ASSERT_EQ(2, state.RejectedPayloads().size());
+}
+
+TEST_F(PkSortedBucketIndexStateTest, RetainsRetiredSourcesButCoversOnlyActiveIntersection) {
+    std::vector<std::shared_ptr<DataFileMeta>> data_files = {
+        MakeDataFile("a", 100, 5, FileSource::Compact()),
+        MakeDataFile("b", 200, 5, FileSource::Compact())};
+    std::shared_ptr<IndexFileMeta> mixed_payload =
+        MakeNamedPayload("mixed.index", 7, "btree", 5, {{"a", 100}, {"retired", 50}}, 150);
+    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<PrimaryKeyIndexSourceFile>{{"a", 100}, {"retired", 50}}),
+              state.Groups()[0]->SourceFiles());
+    ASSERT_EQ((std::vector<PrimaryKeyIndexSourceFile>{{"a", 100}}), state.CoveredSourceFiles());
+    ASSERT_EQ((std::vector<PrimaryKeyIndexSourceFile>{{"b", 200}}), state.UncoveredSourceFiles());
+    ASSERT_TRUE(state.RejectedPayloads().empty());
+}
+
+TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWhoseSourcesAreAllRetired) {
+    std::vector<std::shared_ptr<DataFileMeta>> data_files = {
+        MakeDataFile("active", 100, 5, FileSource::Compact())};
+    std::shared_ptr<IndexFileMeta> 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_TRUE(state.CoveredSourceFiles().empty());
-    ASSERT_EQ(2, state.UncoveredSourceFiles().size());
+    ASSERT_EQ((std::vector<PrimaryKeyIndexSourceFile>{{"active", 100}}),
+              state.UncoveredSourceFiles());
+    ASSERT_EQ((std::vector<std::shared_ptr<IndexFileMeta>>{retired_payload}),
+              state.RejectedPayloads());
+}
+
+TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadForDifferentActiveDataLevel) {
+    std::vector<std::shared_ptr<DataFileMeta>> data_files = {
+        MakeDataFile("active", 100, 4, FileSource::Compact())};
+    std::shared_ptr<IndexFileMeta> wrong_level_payload =
+        MakePayload(7, "btree", 5, {{"active", 100}}, 100);
+    PkSortedBucketIndexState state = PkSortedBucketIndexState::FromActiveDataFiles(
+        7, "btree", data_files, {wrong_level_payload});
+    ASSERT_TRUE(state.Groups().empty());
+    ASSERT_TRUE(state.CoveredSourceFiles().empty());
+    ASSERT_EQ((std::vector<PrimaryKeyIndexSourceFile>{{"active", 100}}),
+              state.UncoveredSourceFiles());
+    ASSERT_EQ((std::vector<std::shared_ptr<IndexFileMeta>>{wrong_level_payload}),
+              state.RejectedPayloads());
 }
 
-TEST_F(PkSortedBucketIndexStateTest, RejectsBothPayloadsWhenLevelHasTwoCandidates) {
+TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadsClaimingTheSameActiveSources) {
     std::vector<std::shared_ptr<DataFileMeta>> data_files = {
         MakeDataFile("a", 100, 5, FileSource::Compact()),
         MakeDataFile("b", 200, 5, FileSource::Compact())};
@@ -268,9 +348,9 @@ TEST_F(PkSortedBucketIndexStateTest, RejectsPayloadWithCorruptSourceMeta) {
     // Version 1 followed by a truncated data level.
     std::shared_ptr<Bytes> corrupt_source_meta =
         std::make_shared<Bytes>(std::string("\x00\x00\x00\x01\x00\x00", 6), pool_.get());
-    std::shared_ptr<IndexFileMeta> payload =
-        MakePayloadWithSourceMetaBytes(7, "btree", /*total_row_count=*/300, /*row_range_start=*/0,
-                                       /*row_range_end=*/299, corrupt_source_meta);
+    std::shared_ptr<IndexFileMeta> 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());