feat(pk-index): maintain BTree indexes during compaction

docs/source/user_guide/primary_key_global_index.rst

@@ -24,11 +24,11 @@ Paimon 2.0 primary-key tables support *source-backed* global scalar indexes
 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
-scan predicate that touches indexed fields against the validated payload groups of the
-scanned snapshot, and narrow covered files to indexed splits carrying file-local row
-ranges. No dedicated query API is required.
+paimon-cpp supports the BTree build, maintenance, and read lifecycle of this protocol for
+fixed-bucket primary-key tables. Compaction automatically builds one immutable payload for
+each indexed field and positive data level. Ordinary batch scans evaluate the indexed part
+of the predicate against validated payload groups and narrow covered files to indexed
+splits carrying file-local row ranges. No dedicated build or query API is required.
 
 Table requirements
 ------------------
@@ -37,7 +37,8 @@ The definitions follow the Java table options:
 
 - ``'pk-btree.index.columns' = 'price'`` with optional
   ``'fields.price.pk-btree.index.options' = '{"block-size":"64 kb"}'``
-- fixed bucket (``bucket > 0``) or postpone bucket mode
+- fixed bucket (``bucket > 0``) for automatic C++ maintenance; Java-compatible postpone
+  bucket schemas remain readable, but the C++ postpone writer does not build payloads
 - ``'deletion-vectors.enabled' = 'true'`` and ``'deletion-vectors.merge-on-read' = 'false'``
 
 Semantics
@@ -50,6 +51,23 @@ Semantics
   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.
+- Index construction reads every physical source row without applying deletion vectors,
+  orders source files by file name, and externally sorts ``(value, group row id)``. During
+  maintenance, missing, duplicate, malformed, or incomplete payloads cause their complete current
+  level to be rebuilt. A payload that still covers every active source is reused even if it also
+  lists retired sources. Data files and the corresponding index ADD / DELETE entries are committed
+  in the same snapshot.
+- The builder uses the existing write-buffer and spill settings. A write context needs a
+  temporary directory when a level exceeds the in-memory write buffer and spill is enabled.
+- If payload construction fails, the data-file transition is still committed. Uncovered files at
+  that level fall back to scanning while any previously usable payload remains active, and a later
+  maintenance attempt can rebuild the complete current source group. Structural commit-increment
+  errors are still rejected.
+- Snapshot expiration retains payloads referenced by the snapshots in its retention set and
+  current-branch live tags, and removes retired payloads before their index manifests, including
+  payloads on an external index path. Expiration is rejected while another branch exists until
+  cross-branch file retention is supported. Orphan cleanup covers table-local index manifests and
+  payloads; it does not enumerate a potentially shared global-index external path.
 - ``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
@@ -70,6 +88,6 @@ Current scope
   currently use the existing C++ length-prefixed UTF-8 streams; ASCII and non-null BMP
   names are compatible with Java ``writeUTF``, while complete modified UTF-8 support for
   supplementary code points will be handled by a shared stream-level change.
-- ``PkSortedIndexFile::Build`` can build one payload for an ordered source group from
-  value-sorted input, which supports tooling and tests; automatic build and maintenance
-  during compaction is not included yet.
+- Automatic maintenance is synchronous during prepare-commit. Java's asynchronous build
+  scheduling, retries, fairness metrics, and manual rebuild actions are not part of the C++
+  API. Realtime and postpone-bucket writers do not build source-backed payloads.

include/paimon/defs.h

@@ -377,6 +377,10 @@ struct PAIMON_EXPORT Options {
     /// a table is never compacted.
     static const char DELETION_VECTORS_ENABLED[];
 
+    /// "pk-clustering-override" - Whether primary-key clustering columns override the primary
+    /// keys when clustering data. Default value is false.
+    static const char PK_CLUSTERING_OVERRIDE[];
+
     /// "deletion-vector.index-file.target-size" - The target size of deletion vector index file.
     /// Default value is 2MB.
     static const char DELETION_VECTOR_INDEX_FILE_TARGET_SIZE[];

include/paimon/orphan_files_cleaner.h

@@ -165,9 +165,9 @@ class PAIMON_EXPORT CleanContextBuilder {
 /// by Paimon C++, we implemented a strong pattern-matching validation, deleting only files in
 /// patterns we recognize.
 ///
-/// @note `OrphanFilesCleaner` in Paimon C++ only support cleaning append table, do not support
-/// cleaning table with tag, table with external paths, table with branch, table with index, table
-/// with changelog, and primary key table.
+/// @note `OrphanFilesCleaner` in Paimon C++ does not support cleaning tables with tags, branches,
+/// external data paths, or changelog manifests. Global-index external paths are not enumerated;
+/// snapshot expiration owns deletion of external index payloads.
 class PAIMON_EXPORT OrphanFilesCleaner {
  public:
     virtual ~OrphanFilesCleaner() = default;

src/paimon/CMakeLists.txt

@@ -263,8 +263,11 @@ set(PAIMON_CORE_SRCS
     core/index/index_file_meta_serializer.cpp
     core/index/pk/primary_key_index_source_meta.cpp
     core/index/pk/primary_key_index_definitions.cpp
+    core/index/pk/bucketed_primary_key_index_maintainer.cpp
     core/index/pksorted/pk_sorted_index_group.cpp
     core/index/pksorted/pk_sorted_bucket_index_state.cpp
+    core/index/pksorted/pk_sorted_data_file_reader.cpp
+    core/index/pksorted/pk_sorted_index_builder.cpp
     core/index/pksorted/pk_sorted_index_file.cpp
     core/io/generic_row_to_arrow_array_converter.cpp
     core/io/meta_to_arrow_array_converter.cpp
@@ -757,6 +760,7 @@ if(PAIMON_BUILD_TESTS)
                     core/index/index_file_meta_serializer_test.cpp
                     core/index/pk/primary_key_index_source_meta_test.cpp
                     core/index/pk/primary_key_index_definitions_test.cpp
+                    core/index/pk/bucketed_primary_key_index_maintainer_test.cpp
                     core/index/pksorted/pk_sorted_bucket_index_state_test.cpp
                     core/index/index_file_handler_test.cpp
                     core/io/compact_increment_test.cpp

src/paimon/common/defs.cpp

@@ -92,6 +92,7 @@ const char Options::FALLBACK_DEDUPLICATE_IGNORE_DELETE[] = "deduplicate.ignore-d
 const char Options::FALLBACK_PARTIAL_UPDATE_IGNORE_DELETE[] = "partial-update.ignore-delete";
 const char Options::FIELDS_DEFAULT_AGG_FUNC[] = "fields.default-aggregate-function";
 const char Options::DELETION_VECTORS_ENABLED[] = "deletion-vectors.enabled";
+const char Options::PK_CLUSTERING_OVERRIDE[] = "pk-clustering-override";
 const char Options::DELETION_VECTOR_INDEX_FILE_TARGET_SIZE[] =
     "deletion-vector.index-file.target-size";
 const char Options::DELETION_VECTOR_BITMAP64[] = "deletion-vectors.bitmap64";

src/paimon/core/index/index_file_handler.cpp

@@ -19,13 +19,30 @@
 
 #include "paimon/core/index/index_file_handler.h"
 
+#include <cstring>
 #include <functional>
 #include <optional>
 
 #include "paimon/core/snapshot.h"
 #include "paimon/status.h"
 
 namespace paimon {
+namespace {
+
+constexpr char kDataEvolutionSourceMetaMagic[] = "DEIX";
+
+}  // namespace
+
+bool IndexFileHandler::IsPrimaryKeySourceIndex(const IndexFileMeta& index_file) {
+    const std::optional<GlobalIndexMeta>& global_index_meta = index_file.GetGlobalIndexMeta();
+    if (!global_index_meta.has_value() || global_index_meta->source_meta == nullptr) {
+        return false;
+    }
+    const std::shared_ptr<Bytes>& source_meta = global_index_meta->source_meta;
+    constexpr size_t kMagicSize = sizeof(kDataEvolutionSourceMetaMagic) - 1;
+    return source_meta->size() < kMagicSize ||
+           std::memcmp(source_meta->data(), kDataEvolutionSourceMetaMagic, kMagicSize) != 0;
+}
 
 Result<IndexFileHandler::IndexFileMetaGroups> IndexFileHandler::Scan(
     const Snapshot& snapshot, const std::string& index_type,
@@ -73,4 +90,20 @@ Result<std::vector<std::shared_ptr<IndexFileMeta>>> IndexFileHandler::Scan(
     return std::vector<std::shared_ptr<IndexFileMeta>>{};
 }
 
+Result<std::vector<std::shared_ptr<IndexFileMeta>>> IndexFileHandler::ScanPrimaryKeyIndexes(
+    const Snapshot& snapshot, const BinaryRow& partition, int32_t bucket) const {
+    std::function<Result<bool>(const IndexManifestEntry&)> filter =
+        [&partition, bucket](const IndexManifestEntry& entry) -> bool {
+        return entry.partition == partition && entry.bucket == bucket &&
+               IsPrimaryKeySourceIndex(*entry.index_file);
+    };
+    PAIMON_ASSIGN_OR_RAISE(std::vector<IndexManifestEntry> entries, Scan(snapshot, filter));
+    std::vector<std::shared_ptr<IndexFileMeta>> result;
+    result.reserve(entries.size());
+    for (const IndexManifestEntry& entry : entries) {
+        result.push_back(entry.index_file);
+    }
+    return result;
+}
+
 }  // namespace paimon

src/paimon/core/index/index_file_handler.h

@@ -55,6 +55,10 @@ class IndexFileHandler {
           dv_bitmap64_(dv_bitmap64),
           pool_(pool) {}
 
+    /// Returns whether an index file carries primary-key source metadata rather than Java's
+    /// data-evolution source metadata.
+    static bool IsPrimaryKeySourceIndex(const IndexFileMeta& index_file);
+
     /// 1.Scan specified index_type index. 2.Cluster with partition & bucket.
     Result<IndexFileMetaGroups> Scan(const Snapshot& snapshot, const std::string& index_type,
                                      const std::unordered_set<BinaryRow>& partitions) const;
@@ -64,6 +68,10 @@ class IndexFileHandler {
                                                              const BinaryRow& partition,
                                                              int32_t bucket) const;
 
+    /// Scan primary-key source-backed index payloads for a partition and bucket.
+    Result<std::vector<std::shared_ptr<IndexFileMeta>>> ScanPrimaryKeyIndexes(
+        const Snapshot& snapshot, const BinaryRow& partition, int32_t bucket) const;
+
     /// Scan specified all typed index.
     Result<std::vector<IndexManifestEntry>> Scan(
         const Snapshot& snapshot,

src/paimon/core/index/index_file_handler_test.cpp

@@ -31,6 +31,7 @@
 #include "paimon/core/core_options.h"
 #include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
 #include "paimon/core/index/deletion_vector_meta.h"
+#include "paimon/core/index/global_index_meta.h"
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/core/snapshot.h"
@@ -288,6 +289,31 @@ TEST_F(IndexFileHandlerTest, TestScanWithNoIndexManifest) {
     ASSERT_TRUE(index_entries.empty());
 }
 
+TEST_F(IndexFileHandlerTest, TestScanPrimaryKeyIndexesBySourceMetadata) {
+    std::string table_path =
+        paimon::test::GetDataDir() + "/orc/pk_btree_source_meta.db/pk_btree_source_meta/";
+    ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
+                         CoreOptions::FromMap({{Options::MANIFEST_FORMAT, "orc"}}));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexFileHandler> index_file_handler,
+                         CreateIndexFileHandler(table_path, core_options));
+
+    SnapshotManager snapshot_manager(core_options.GetFileSystem(), table_path);
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, snapshot_manager.LoadSnapshot(/*snapshot_id=*/5));
+    ASSERT_OK_AND_ASSIGN(
+        std::vector<std::shared_ptr<IndexFileMeta>> source_indexes,
+        index_file_handler->ScanPrimaryKeyIndexes(snapshot, BinaryRow::EmptyRow(), /*bucket=*/0));
+
+    ASSERT_EQ(source_indexes.size(), 1);
+    const std::optional<GlobalIndexMeta>& global_index_meta =
+        source_indexes[0]->GetGlobalIndexMeta();
+    ASSERT_TRUE(global_index_meta.has_value());
+    ASSERT_NE(global_index_meta->source_meta, nullptr);
+
+    ASSERT_OK_AND_ASSIGN(source_indexes, index_file_handler->ScanPrimaryKeyIndexes(
+                                             snapshot, BinaryRow::EmptyRow(), /*bucket=*/1));
+    ASSERT_TRUE(source_indexes.empty());
+}
+
 TEST_F(IndexFileHandlerTest, TestScanByPartitionBucketAndReadAllDeletionVectors) {
     std::string table_path = paimon::test::GetDataDir() +
                              "/orc/pk_table_with_dv_cardinality.db/pk_table_with_dv_cardinality/";