feat(core): read source-backed primary-key BTree indexes

docs/source/user_guide.rst

@@ -42,3 +42,4 @@ User Guide
    user_guide/prefetch
    user_guide/arrow
    user_guide/global_index
+   user_guide/primary_key_global_index

docs/source/user_guide/primary_key_global_index.rst

@@ -0,0 +1,73 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements.  See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership.  The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License.  You may obtain a copy of the License at
+
+..   http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied.  See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+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.
+
+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.
+
+Table requirements
+------------------
+
+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
+- ``'deletion-vectors.enabled' = 'true'`` and ``'deletion-vectors.merge-on-read' = 'false'``
+
+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.
+- ``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
+  normal scan individually.
+- Indexed splits keep their deletion files aligned with the data file; the reader still
+  applies deletion vectors and the complete original predicate, so index results never
+  change visibility semantics.
+- ``'global-index.enabled' = 'false'`` disables the planner.
+
+Current scope
+-------------
+
+- The BTree payload reader is wired up. ``pk-bitmap`` (and vector / full-text)
+  definitions are recognized for validation, but their evaluation conservatively falls
+  back to a normal scan until their dedicated payload readers are supported.
+- The read path targets the Java release-2.0.0 layout and scan semantics (source metadata
+  v1, ``GlobalIndexMeta`` with ``_SOURCE_META``, commit message v12). Source-file names
+  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.

include/paimon/defs.h

@@ -522,6 +522,18 @@ struct PAIMON_EXPORT Options {
     /// "global-index.external-path" - Global index root directory, if not set, the global index
     /// files will be stored under the index directory.
     static const char GLOBAL_INDEX_EXTERNAL_PATH[];
+    /// "pk-btree.index.columns" - Comma-separated columns indexed by primary-key BTree indexes.
+    /// No default value.
+    static const char PK_BTREE_INDEX_COLUMNS[];
+    /// "pk-bitmap.index.columns" - Comma-separated columns indexed by primary-key Bitmap indexes.
+    /// No default value.
+    static const char PK_BITMAP_INDEX_COLUMNS[];
+    /// "pk-vector.index.columns" - Comma-separated VECTOR columns indexed by primary-key vector
+    /// indexes. No default value.
+    static const char PK_VECTOR_INDEX_COLUMNS[];
+    /// "pk-full-text.index.columns" - Comma-separated character columns indexed by primary-key
+    /// full-text indexes. No default value.
+    static const char PK_FULL_TEXT_INDEX_COLUMNS[];
     /// "aggregation.remove-record-on-delete" - Whether to remove the whole row in aggregation
     /// engine when delete records are received. Default value is "false".
     static const char AGGREGATION_REMOVE_RECORD_ON_DELETE[];

src/paimon/CMakeLists.txt

@@ -254,6 +254,11 @@ set(PAIMON_CORE_SRCS
     core/index/index_file_handler.cpp
     core/index/global_index_meta.cpp
     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/pksorted/pk_sorted_index_group.cpp
+    core/index/pksorted/pk_sorted_bucket_index_state.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
     core/io/async_key_value_producer_and_consumer.cpp
@@ -405,6 +410,9 @@ set(PAIMON_CORE_SRCS
     core/table/source/table_read.cpp
     core/table/source/table_scan.cpp
     core/table/source/data_evolution_batch_scan.cpp
+    core/table/source/primary_key_sorted_index_scan.cpp
+    core/table/source/primary_key_sorted_index_result.cpp
+    core/table/source/primary_key_index_batch_scan.cpp
     core/table/system/audit_log_system_table.cpp
     core/table/system/binlog_system_table.cpp
     core/table/system/global_system_tables.cpp
@@ -723,6 +731,9 @@ if(PAIMON_BUILD_TESTS)
                     core/index/index_in_data_file_dir_path_factory_test.cpp
                     core/index/deletion_vector_meta_test.cpp
                     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/pksorted/pk_sorted_bucket_index_state_test.cpp
                     core/index/index_file_handler_test.cpp
                     core/io/compact_increment_test.cpp
                     core/io/infer_shredding_file_writer_test.cpp
@@ -740,6 +751,7 @@ if(PAIMON_BUILD_TESTS)
                     core/io/file_index_evaluator_test.cpp
                     core/io/single_file_writer_test.cpp
                     core/io/rolling_blob_file_writer_test.cpp
+                    core/global_index/global_index_evaluator_impl_test.cpp
                     core/global_index/indexed_split_test.cpp
                     core/manifest/file_source_test.cpp
                     core/manifest/file_kind_test.cpp
@@ -859,6 +871,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/sink/commit_message_test.cpp
                     core/table/sink/commit_message_impl_test.cpp
                     core/table/source/fallback_data_split_test.cpp
+                    core/table/source/primary_key_sorted_index_scan_test.cpp
                     core/table/source/table_read_test.cpp
                     core/table/source/append_count_reader_test.cpp
                     core/table/source/pk_count_reader_test.cpp

src/paimon/common/defs.cpp

@@ -139,6 +139,10 @@ const char Options::BLOB_WRITE_NULL_ON_FETCH_FAILURE[] = "blob-write-null-on-fet
 const char Options::GLOBAL_INDEX_ENABLED[] = "global-index.enabled";
 const char Options::GLOBAL_INDEX_THREAD_NUM[] = "global-index.thread-num";
 const char Options::GLOBAL_INDEX_EXTERNAL_PATH[] = "global-index.external-path";
+const char Options::PK_BTREE_INDEX_COLUMNS[] = "pk-btree.index.columns";
+const char Options::PK_BITMAP_INDEX_COLUMNS[] = "pk-bitmap.index.columns";
+const char Options::PK_VECTOR_INDEX_COLUMNS[] = "pk-vector.index.columns";
+const char Options::PK_FULL_TEXT_INDEX_COLUMNS[] = "pk-full-text.index.columns";
 const char Options::AGGREGATION_REMOVE_RECORD_ON_DELETE[] = "aggregation.remove-record-on-delete";
 const char Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED[] = "table-read.sequence-number.enabled";
 const char Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED[] = "key-value.sequence_number.enabled";

src/paimon/common/global_index/btree/btree_compatibility_test.cpp

@@ -783,7 +783,8 @@ TEST_F(BTreeCompatibilityTest, MetaDeserialization) {
         auto meta_str = ReadFileAsString(meta_path);
         std::shared_ptr<Bytes> meta_bytes = Bytes::AllocateBytes(meta_str, pool_.get());
 
-        auto meta = BTreeIndexMeta::Deserialize(meta_bytes, pool_.get());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BTreeIndexMeta> meta,
+                             BTreeIndexMeta::Deserialize(meta_bytes, pool_.get()));
         ASSERT_TRUE(meta);
 
         ASSERT_TRUE(meta->HasNulls());
@@ -808,7 +809,8 @@ TEST_F(BTreeCompatibilityTest, MetaDeserialization) {
         auto meta_str = ReadFileAsString(meta_path);
         std::shared_ptr<Bytes> meta_bytes = Bytes::AllocateBytes(meta_str, pool_.get());
 
-        auto meta = BTreeIndexMeta::Deserialize(meta_bytes, pool_.get());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BTreeIndexMeta> meta,
+                             BTreeIndexMeta::Deserialize(meta_bytes, pool_.get()));
         ASSERT_TRUE(meta);
 
         ASSERT_TRUE(meta->HasNulls());
@@ -833,7 +835,8 @@ TEST_F(BTreeCompatibilityTest, MetaDeserialization) {
         auto meta_str = ReadFileAsString(meta_path);
         std::shared_ptr<Bytes> meta_bytes = Bytes::AllocateBytes(meta_str, pool_.get());
 
-        auto meta = BTreeIndexMeta::Deserialize(meta_bytes, pool_.get());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BTreeIndexMeta> meta,
+                             BTreeIndexMeta::Deserialize(meta_bytes, pool_.get()));
         ASSERT_TRUE(meta);
 
         ASSERT_TRUE(meta->HasNulls());
@@ -848,7 +851,8 @@ TEST_F(BTreeCompatibilityTest, MetaDeserialization) {
         auto meta_str = ReadFileAsString(meta_path);
         std::shared_ptr<Bytes> meta_bytes = Bytes::AllocateBytes(meta_str, pool_.get());
 
-        auto meta = BTreeIndexMeta::Deserialize(meta_bytes, pool_.get());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BTreeIndexMeta> meta,
+                             BTreeIndexMeta::Deserialize(meta_bytes, pool_.get()));
         ASSERT_TRUE(meta);
 
         ASSERT_TRUE(meta->FirstKey());
@@ -870,7 +874,8 @@ TEST_F(BTreeCompatibilityTest, MetaDeserialization) {
         auto meta_str = ReadFileAsString(meta_path);
         std::shared_ptr<Bytes> meta_bytes = Bytes::AllocateBytes(meta_str, pool_.get());
 
-        auto meta = BTreeIndexMeta::Deserialize(meta_bytes, pool_.get());
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<BTreeIndexMeta> meta,
+                             BTreeIndexMeta::Deserialize(meta_bytes, pool_.get()));
         ASSERT_TRUE(meta);
 
         ASSERT_TRUE(meta->HasNulls());

src/paimon/common/global_index/btree/btree_file_meta_selector.cpp

@@ -19,22 +19,70 @@
 
 #include "paimon/common/global_index/btree/btree_file_meta_selector.h"
 
+#include "fmt/format.h"
 #include "paimon/common/memory/memory_slice.h"
 
 namespace paimon {
-BTreeFileMetaSelector::BTreeFileMetaSelector(const std::vector<GlobalIndexIOMeta>& files,
-                                             const std::shared_ptr<arrow::DataType>& key_type,
-                                             const std::shared_ptr<MemoryPool>& pool)
-    : key_type_(key_type),
-      pool_(pool),
-      comparator_(KeySerializer::CreateComparator(key_type, pool)) {
-    files_.reserve(files.size());
+Result<std::unique_ptr<BTreeFileMetaSelector>> BTreeFileMetaSelector::Create(
+    const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<arrow::DataType>& key_type,
+    const std::shared_ptr<MemoryPool>& pool) {
+    if (key_type == nullptr) {
+        return Status::Invalid("Cannot create a BTree file metadata selector without a key type.");
+    }
+    if (pool == nullptr) {
+        return Status::Invalid(
+            "Cannot create a BTree file metadata selector without a memory pool.");
+    }
+    std::vector<std::pair<GlobalIndexIOMeta, std::shared_ptr<BTreeIndexMeta>>> decoded_files;
+    decoded_files.reserve(files.size());
+    MemorySlice::SliceComparator comparator = KeySerializer::CreateComparator(key_type, pool);
     for (const auto& file : files) {
-        auto index_meta = BTreeIndexMeta::Deserialize(file.metadata, pool.get());
-        files_.emplace_back(file, std::move(index_meta));
+        PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BTreeIndexMeta> index_meta,
+                               BTreeIndexMeta::Deserialize(file.metadata, pool.get()));
+        bool has_first_key = index_meta->FirstKey() != nullptr;
+        bool has_last_key = index_meta->LastKey() != nullptr;
+        if (has_first_key != has_last_key) {
+            return Status::Invalid(fmt::format(
+                "BTree index metadata for {} must contain both boundary keys or neither.",
+                file.file_path));
+        }
+        if (!has_first_key && !index_meta->HasNulls()) {
+            return Status::Invalid(
+                fmt::format("BTree index metadata for {} has no boundary keys or null values.",
+                            file.file_path));
+        }
+        if (index_meta->FirstKey() != nullptr) {
+            PAIMON_RETURN_NOT_OK(KeySerializer::ValidateSerializedKey(
+                WrapKeySlice(index_meta->FirstKey()), key_type));
+        }
+        if (index_meta->LastKey() != nullptr) {
+            PAIMON_RETURN_NOT_OK(KeySerializer::ValidateSerializedKey(
+                WrapKeySlice(index_meta->LastKey()), key_type));
+        }
+        if (index_meta->FirstKey() != nullptr && index_meta->LastKey() != nullptr) {
+            PAIMON_ASSIGN_OR_RAISE(int32_t comparison,
+                                   comparator(WrapKeySlice(index_meta->FirstKey()),
+                                              WrapKeySlice(index_meta->LastKey())));
+            if (comparison > 0) {
+                return Status::Invalid(fmt::format(
+                    "BTree index metadata for {} has a first key greater than its last key.",
+                    file.file_path));
+            }
+        }
+        decoded_files.emplace_back(file, std::move(index_meta));
     }
+    return std::unique_ptr<BTreeFileMetaSelector>(
+        new BTreeFileMetaSelector(std::move(decoded_files), key_type, pool));
 }
 
+BTreeFileMetaSelector::BTreeFileMetaSelector(
+    std::vector<std::pair<GlobalIndexIOMeta, std::shared_ptr<BTreeIndexMeta>>> files,
+    std::shared_ptr<arrow::DataType> key_type, std::shared_ptr<MemoryPool> pool)
+    : files_(std::move(files)),
+      key_type_(std::move(key_type)),
+      pool_(std::move(pool)),
+      comparator_(KeySerializer::CreateComparator(key_type_, pool_)) {}
+
 Result<std::vector<GlobalIndexIOMeta>> BTreeFileMetaSelector::VisitIsNotNull() {
     return Filter([](const BTreeIndexMeta& meta) -> Result<bool> { return !meta.OnlyNulls(); });
 }
@@ -208,7 +256,9 @@ MemorySlice BTreeFileMetaSelector::WrapKeySlice(const std::shared_ptr<Bytes>& ke
 Result<MemorySlice> BTreeFileMetaSelector::SerializeLiteral(const Literal& literal) const {
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes,
                            KeySerializer::SerializeKey(literal, key_type_, pool_.get()));
-    return MemorySlice::Wrap(bytes);
+    MemorySlice slice = MemorySlice::Wrap(bytes);
+    PAIMON_RETURN_NOT_OK(KeySerializer::ValidateSerializedKey(slice, key_type_));
+    return slice;
 }
 
 }  // namespace paimon

src/paimon/common/global_index/btree/btree_file_meta_selector.h

@@ -34,9 +34,9 @@ namespace paimon {
 /// Selects candidate BTree index files based on filter predicates.
 class BTreeFileMetaSelector : public FunctionVisitor<std::vector<GlobalIndexIOMeta>> {
  public:
-    BTreeFileMetaSelector(const std::vector<GlobalIndexIOMeta>& files,
-                          const std::shared_ptr<arrow::DataType>& key_type,
-                          const std::shared_ptr<MemoryPool>& pool);
+    static Result<std::unique_ptr<BTreeFileMetaSelector>> Create(
+        const std::vector<GlobalIndexIOMeta>& files,
+        const std::shared_ptr<arrow::DataType>& key_type, const std::shared_ptr<MemoryPool>& pool);
 
     Result<std::vector<GlobalIndexIOMeta>> VisitIsNotNull() override;
     Result<std::vector<GlobalIndexIOMeta>> VisitIsNull() override;
@@ -55,6 +55,10 @@ class BTreeFileMetaSelector : public FunctionVisitor<std::vector<GlobalIndexIOMe
     Result<std::vector<GlobalIndexIOMeta>> VisitLike(const Literal& literal) override;
 
  private:
+    BTreeFileMetaSelector(
+        std::vector<std::pair<GlobalIndexIOMeta, std::shared_ptr<BTreeIndexMeta>>> files,
+        std::shared_ptr<arrow::DataType> key_type, std::shared_ptr<MemoryPool> pool);
+
     using MetaPredicate = std::function<Result<bool>(const BTreeIndexMeta&)>;
 
     Result<std::vector<GlobalIndexIOMeta>> Filter(const MetaPredicate& predicate) const;