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[];

include/paimon/global_index/global_indexer.h

@@ -35,6 +35,8 @@
 struct ArrowSchema;
 
 namespace paimon {
+class Executor;
+
 /// Interface for creating global index readers and writers.
 class PAIMON_EXPORT GlobalIndexer {
  public:
@@ -70,6 +72,27 @@ class PAIMON_EXPORT GlobalIndexer {
         ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
         const std::vector<GlobalIndexIOMeta>& files,
         const std::shared_ptr<MemoryPool>& pool) const = 0;
+
+    /// Creates a reader using an executor supplied by the scan layer.
+    ///
+    /// Index implementations which do not perform asynchronous work may ignore the executor and
+    /// use the compatibility overload above.
+    ///
+    /// @param arrow_schema   Schema of the indexed data; used to interpret predicate literals.
+    /// @param file_reader    I/O handler for reading index artifacts from storage.
+    /// @param files          List of index file metadata entries produced during writing.
+    /// @param pool           Memory pool for temporary allocations; if nullptr, uses default.
+    /// @param executor       Executor shared by readers created for the same scan; nullptr means
+    ///                       that the reader should evaluate sequentially.
+    /// @return A `Result` containing a shared pointer to the created `GlobalIndexReader`,
+    ///         or an error if the index cannot be loaded or is incompatible, etc.
+    virtual Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
+        ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
+        const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<MemoryPool>& pool,
+        const std::shared_ptr<Executor>& executor) const {
+        static_cast<void>(executor);
+        return CreateReader(arrow_schema, file_reader, files, pool);
+    }
 };
 
 }  // namespace paimon

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
@@ -404,6 +409,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
@@ -722,6 +730,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
@@ -857,6 +868,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_global_index_integration_test.cpp

@@ -16,6 +16,10 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+#include <atomic>
+#include <cstdint>
+#include <functional>
+
 #include "arrow/c/bridge.h"
 #include "arrow/ipc/json_simple.h"
 #include "gtest/gtest.h"
@@ -29,6 +33,7 @@
 #include "paimon/common/utils/scope_guard.h"
 #include "paimon/data/decimal.h"
 #include "paimon/data/timestamp.h"
+#include "paimon/executor.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
 #include "paimon/global_index/io/global_index_file_reader.h"
@@ -84,6 +89,27 @@ class FakeGlobalIndexFileReader : public GlobalIndexFileReader {
     std::string base_path_;
 };
 
+class CountingInlineExecutor : public Executor {
+ public:
+    void Add(std::function<void()> func) override {
+        submission_count_.fetch_add(1);
+        func();
+    }
+
+    void ShutdownNow() override {}
+
+    uint32_t GetThreadNum() const override {
+        return 1;
+    }
+
+    uint32_t SubmissionCount() const {
+        return submission_count_.load();
+    }
+
+ private:
+    std::atomic<uint32_t> submission_count_{0};
+};
+
 class BTreeGlobalIndexIntegrationTest : public ::testing::Test,
                                         public ::testing::WithParamInterface<std::string> {
  protected:
@@ -1972,9 +1998,10 @@ TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadMultiFilesWithMetaSelector)
     // Create reader over all 3 files (internally uses LazyFilteredBTreeReader +
     // BTreeFileMetaSelector)
     auto file_reader = std::make_shared<FakeGlobalIndexFileReader>(fs_, base_path_);
+    auto executor = std::make_shared<CountingInlineExecutor>();
     auto c_schema = CreateArrowSchema(field);
-    ASSERT_OK_AND_ASSIGN(auto reader,
-                         indexer->CreateReader(c_schema.get(), file_reader, all_metas, pool_));
+    ASSERT_OK_AND_ASSIGN(auto reader, indexer->CreateReader(c_schema.get(), file_reader, all_metas,
+                                                            pool_, executor));
 
     // --- VisitEqual: key=12 -> only file1 is selected by meta selector -> row 5
     {
@@ -2023,6 +2050,7 @@ TEST_P(BTreeGlobalIndexIntegrationTest, WriteAndReadMultiFilesWithMetaSelector)
         Literal literal_5(5);
         ASSERT_OK_AND_ASSIGN(auto result, reader->VisitGreaterOrEqual(literal_5));
         CheckResult(result, {3, 4, 5, 6, 7, 9});
+        ASSERT_EQ(executor->SubmissionCount(), 3);
     }
 
     // --- VisitLessOrEqual: key <= 2 -> only file0 selected -> rows 0,1

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

@@ -18,10 +18,12 @@
  */
 #include "paimon/common/global_index/btree/btree_global_indexer.h"
 
+#include <climits>
 #include <memory>
 #include <string>
 
 #include "arrow/c/bridge.h"
+#include "fmt/format.h"
 #include "paimon/common/compression/block_compression_factory.h"
 #include "paimon/common/global_index/btree/btree_file_footer.h"
 #include "paimon/common/global_index/btree/btree_global_index_writer.h"
@@ -36,10 +38,10 @@
 #include "paimon/common/utils/options_utils.h"
 #include "paimon/common/utils/preconditions.h"
 #include "paimon/core/options/compress_options.h"
-#include "paimon/executor.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
 #include "paimon/memory/bytes.h"
 #include "paimon/utils/roaring_bitmap64.h"
+
 namespace paimon {
 Result<std::unique_ptr<BTreeGlobalIndexer>> BTreeGlobalIndexer::Create(
     const std::map<std::string, std::string>& options) {
@@ -100,6 +102,13 @@ Result<std::shared_ptr<GlobalIndexWriter>> BTreeGlobalIndexer::CreateWriter(
 Result<std::shared_ptr<GlobalIndexReader>> BTreeGlobalIndexer::CreateReader(
     ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
     const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<MemoryPool>& pool) const {
+    return CreateReader(arrow_schema, file_reader, files, pool, /*executor=*/nullptr);
+}
+
+Result<std::shared_ptr<GlobalIndexReader>> BTreeGlobalIndexer::CreateReader(
+    ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
+    const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<MemoryPool>& pool,
+    const std::shared_ptr<Executor>& executor) const {
     // Get field type from arrow schema
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> schema,
                                       arrow::ImportSchema(arrow_schema));
@@ -120,8 +129,6 @@ Result<std::shared_ptr<GlobalIndexReader>> BTreeGlobalIndexer::CreateReader(
         }
         read_buffer_size = static_cast<int32_t>(tmp_buffer_size);
     }
-    // TODO(lisizhuo.lsz): Allow users to specify an executor
-    std::shared_ptr<Executor> executor = CreateDefaultExecutor();
     return std::make_shared<LazyFilteredBTreeReader>(read_buffer_size, files, key_type, file_reader,
                                                      cache_manager_, pool, executor);
 }

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

@@ -68,6 +68,11 @@ class BTreeGlobalIndexer : public GlobalIndexer {
         const std::vector<GlobalIndexIOMeta>& files,
         const std::shared_ptr<MemoryPool>& pool) const override;
 
+    Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
+        ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
+        const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<MemoryPool>& pool,
+        const std::shared_ptr<Executor>& executor) const override;
+
  private:
     BTreeGlobalIndexer(const std::shared_ptr<CacheManager>& cache_manager,
                        const std::map<std::string, std::string>& options)

src/paimon/core/index/pk/primary_key_index_definition.h

@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <map>
+#include <string>
+#include <utility>
+
+namespace paimon {
+/// Resolved definition of one source-backed primary-key index.
+class PrimaryKeyIndexDefinition {
+ public:
+    /// Built-in primary-key index families.
+    enum class Family {
+        VECTOR,
+        BTREE,
+        BITMAP,
+        FULL_TEXT,
+    };
+
+    PrimaryKeyIndexDefinition(std::string column, int32_t field_id, std::string index_type,
+                              Family family, std::map<std::string, std::string> options)
+        : column_(std::move(column)),
+          field_id_(field_id),
+          index_type_(std::move(index_type)),
+          family_(family),
+          options_(std::move(options)) {}
+
+    const std::string& Column() const {
+        return column_;
+    }
+
+    int32_t FieldId() const {
+        return field_id_;
+    }
+
+    const std::string& IndexType() const {
+        return index_type_;
+    }
+
+    Family GetFamily() const {
+        return family_;
+    }
+
+    const std::map<std::string, std::string>& Options() const {
+        return options_;
+    }
+
+ private:
+    std::string column_;
+    int32_t field_id_;
+    std::string index_type_;
+    Family family_;
+    std::map<std::string, std::string> options_;
+};
+
+}  // namespace paimon