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 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.
 
-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
@@ -48,6 +49,21 @@ Semantics
   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.
+- Index construction reads every physical source row without applying deletion vectors,
+  orders source files by file name, and externally sorts ``(value, group row id)``. Missing,
+  duplicate, malformed, or stale payloads cause their complete level to be rebuilt. 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 and the affected
+  level remains uncovered; scans fall back to the data files and a later maintenance attempt can
+  rebuild the payload. 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
@@ -68,6 +84,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

@@ -369,6 +369,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

@@ -262,8 +262,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
@@ -750,6 +753,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

@@ -90,6 +90,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/common/utils/fields_comparator.cpp

@@ -47,11 +47,26 @@ Result<std::unique_ptr<FieldsComparator>> FieldsComparator::Create(
 Result<std::unique_ptr<FieldsComparator>> FieldsComparator::Create(
     const std::vector<DataField>& input_data_field, const std::vector<int32_t>& sort_fields,
     bool is_ascending_order) {
+    return Create(input_data_field, sort_fields, is_ascending_order,
+                  /*use_java_floating_point_order=*/false);
+}
+
+Result<std::unique_ptr<FieldsComparator>> FieldsComparator::CreateWithJavaFloatingPointOrder(
+    const std::vector<DataField>& input_data_field, const std::vector<int32_t>& sort_fields,
+    bool is_ascending_order) {
+    return Create(input_data_field, sort_fields, is_ascending_order,
+                  /*use_java_floating_point_order=*/true);
+}
+
+Result<std::unique_ptr<FieldsComparator>> FieldsComparator::Create(
+    const std::vector<DataField>& input_data_field, const std::vector<int32_t>& sort_fields,
+    bool is_ascending_order, bool use_java_floating_point_order) {
     std::vector<FieldComparatorFunc> comparators;
     comparators.reserve(sort_fields.size());
     for (const auto& sort_field_idx : sort_fields) {
         const auto& type = input_data_field[sort_field_idx].Type();
-        PAIMON_ASSIGN_OR_RAISE(FieldComparatorFunc cmp, CompareField(sort_field_idx, type));
+        PAIMON_ASSIGN_OR_RAISE(FieldComparatorFunc cmp,
+                               CompareField(sort_field_idx, type, use_java_floating_point_order));
         comparators.emplace_back(cmp);
     }
     return std::unique_ptr<FieldsComparator>(
@@ -81,7 +96,8 @@ int32_t FieldsComparator::CompareTo(const InternalRow& lhs, const InternalRow& r
 }
 
 Result<FieldsComparator::FieldComparatorFunc> FieldsComparator::CompareField(
-    int32_t field_idx, const std::shared_ptr<arrow::DataType>& input_type) {
+    int32_t field_idx, const std::shared_ptr<arrow::DataType>& input_type,
+    bool use_java_floating_point_order) {
     arrow::Type::type type = input_type->id();
     switch (type) {
         case arrow::Type::type::BOOL:
@@ -128,21 +144,26 @@ Result<FieldsComparator::FieldComparatorFunc> FieldsComparator::CompareField(
                     return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
                 });
         case arrow::Type::type::FLOAT:
-            // TODO(xinyu.lxy):
-            // currently in java KeyComparatorSupplier: -inf < -0.0 == +0.0 < +inf = nan
-            // paimon-cpp: -inf < -0.0 == +0.0 < +inf and nan cannot be compared
+            // The legacy branch does not define a strict order for NaN. Primary-key BTree index
+            // construction opts into the Java floating-point order instead.
             return FieldsComparator::FieldComparatorFunc(
-                [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
+                [field_idx, use_java_floating_point_order](const InternalRow& lhs,
+                                                           const InternalRow& rhs) -> int32_t {
                     float lvalue = lhs.GetFloat(field_idx);
                     float rvalue = rhs.GetFloat(field_idx);
-                    return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
+                    return use_java_floating_point_order
+                               ? CompareFloatingPoint(lvalue, rvalue)
+                               : (lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1));
                 });
         case arrow::Type::type::DOUBLE:
             return FieldsComparator::FieldComparatorFunc(
-                [field_idx](const InternalRow& lhs, const InternalRow& rhs) -> int32_t {
+                [field_idx, use_java_floating_point_order](const InternalRow& lhs,
+                                                           const InternalRow& rhs) -> int32_t {
                     double lvalue = lhs.GetDouble(field_idx);
                     double rvalue = rhs.GetDouble(field_idx);
-                    return lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1);
+                    return use_java_floating_point_order
+                               ? CompareFloatingPoint(lvalue, rvalue)
+                               : (lvalue == rvalue ? 0 : (lvalue < rvalue ? -1 : 1));
                 });
         case arrow::Type::type::STRING:
         case arrow::Type::type::BINARY: {

src/paimon/common/utils/fields_comparator.h

@@ -20,6 +20,7 @@
 #pragma once
 
 #include <cassert>
+#include <cmath>
 #include <cstdint>
 #include <functional>
 #include <memory>
@@ -48,6 +49,11 @@ class FieldsComparator {
         const std::vector<DataField>& input_data_field, const std::vector<int32_t>& sort_fields,
         bool is_ascending_order);
 
+    /// Create a comparator with Java's floating-point total order for source-backed indexes.
+    static Result<std::unique_ptr<FieldsComparator>> CreateWithJavaFloatingPointOrder(
+        const std::vector<DataField>& input_data_field, const std::vector<int32_t>& sort_fields,
+        bool is_ascending_order);
+
     int32_t CompareTo(const InternalRow& lhs, const InternalRow& rhs) const;
 
     const std::vector<int32_t>& CompareFields() const {
@@ -93,8 +99,13 @@ class FieldsComparator {
         assert(comparators_.size() == sort_fields_.size());
     }
 
+    static Result<std::unique_ptr<FieldsComparator>> Create(
+        const std::vector<DataField>& input_data_field, const std::vector<int32_t>& sort_fields,
+        bool is_ascending_order, bool use_java_floating_point_order);
+
     static Result<FieldComparatorFunc> CompareField(
-        int32_t field_idx, const std::shared_ptr<arrow::DataType>& input_type);
+        int32_t field_idx, const std::shared_ptr<arrow::DataType>& input_type,
+        bool use_java_floating_point_order);
 
  private:
     bool is_ascending_order_;

src/paimon/common/utils/fields_comparator_test.cpp

@@ -20,6 +20,7 @@
 #include "paimon/common/utils/fields_comparator.h"
 
 #include <cstddef>
+#include <limits>
 #include <string>
 #include <variant>
 
@@ -80,6 +81,29 @@ class FieldsComparatorTest : public ::testing::Test {
         }
         CheckResult(row1, row2, input_types, sort_fields, has_null);
     }
+
+    void CheckJavaFloatingPointResult(
+        const InternalRow& row1, const InternalRow& row2,
+        const std::vector<std::shared_ptr<arrow::DataType>>& input_types) {
+        std::vector<DataField> data_fields;
+        data_fields.reserve(input_types.size());
+        for (int32_t i = 0; i < static_cast<int32_t>(input_types.size()); ++i) {
+            data_fields.emplace_back(i, arrow::field("fake_name", input_types[i]));
+        }
+        for (int32_t i = 0; i < static_cast<int32_t>(input_types.size()); ++i) {
+            ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldsComparator> ascending,
+                                 FieldsComparator::CreateWithJavaFloatingPointOrder(
+                                     data_fields, {i}, /*is_ascending_order=*/true));
+            ASSERT_EQ(-1, ascending->CompareTo(row1, row2));
+            ASSERT_EQ(1, ascending->CompareTo(row2, row1));
+
+            ASSERT_OK_AND_ASSIGN(std::unique_ptr<FieldsComparator> descending,
+                                 FieldsComparator::CreateWithJavaFloatingPointOrder(
+                                     data_fields, {i}, /*is_ascending_order=*/false));
+            ASSERT_EQ(1, descending->CompareTo(row1, row2));
+            ASSERT_EQ(-1, descending->CompareTo(row2, row1));
+        }
+    }
 };
 
 TEST_F(FieldsComparatorTest, TestSimple) {
@@ -202,6 +226,26 @@ TEST_F(FieldsComparatorTest, TestSimple) {
     }
 }
 
+TEST_F(FieldsComparatorTest, TestFloatingPointTotalOrder) {
+    auto pool = GetDefaultPool();
+    BinaryRow negative_zero = BinaryRowGenerator::GenerateRow({-0.0F, -0.0}, pool.get());
+    BinaryRow positive_zero = BinaryRowGenerator::GenerateRow({0.0F, 0.0}, pool.get());
+    CheckJavaFloatingPointResult(negative_zero, positive_zero,
+                                 {arrow::float32(), arrow::float64()});
+    ASSERT_OK_AND_ASSIGN(
+        std::unique_ptr<FieldsComparator> default_comparator,
+        FieldsComparator::Create({DataField(0, arrow::field("float", arrow::float32())),
+                                  DataField(1, arrow::field("double", arrow::float64()))},
+                                 /*is_ascending_order=*/true));
+    ASSERT_EQ(0, default_comparator->CompareTo(negative_zero, positive_zero));
+
+    const float float_nan = std::numeric_limits<float>::quiet_NaN();
+    const double double_nan = std::numeric_limits<double>::quiet_NaN();
+    BinaryRow finite = BinaryRowGenerator::GenerateRow({1.0F, 1.0}, pool.get());
+    BinaryRow nan = BinaryRowGenerator::GenerateRow({float_nan, double_nan}, pool.get());
+    CheckJavaFloatingPointResult(finite, nan, {arrow::float32(), arrow::float64()});
+}
+
 TEST_F(FieldsComparatorTest, TestTimestampType) {
     auto pool = GetDefaultPool();
     // test ts with different precision

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,