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

[wangyong9999]

Purpose

Linked issue: Closes #192

This PR adds C++ read-side support for source-backed primary-key BTree indexes produced by Apache Paimon Java 2.0, together with an internal payload builder used by tooling and tests.

The main changes are:

• Decode and serialize PrimaryKeyIndexSourceMeta v1 using its big-endian layout and length-prefixed file-name bytes, with validation for unsupported versions, invalid counts, truncation, and trailing bytes.
• Parse the pk-btree, pk-bitmap, pk-vector, and pk-full-text column definitions, including field-scoped JSON options and duplicate or cross-family assignment checks.
• Validate index payload coverage against the active ordered COMPACT files of each positive data level, including source names, order, row counts, field ID, index type, complete row range, and known zero delete-row counts.
• Decode BTree file metadata through checked Result APIs and reject missing, truncated, type-invalid, one-sided, empty non-null, or reversed boundaries before pruning.
• Normalize predicates in GlobalIndexEvaluatorImpl and evaluate indexed predicates conservatively: AND may use safely evaluable children, while OR requires every branch to be evaluable.
• Evaluate each validated BTree group once, localize group ordinals to file-local physical row ranges, and produce IndexedSplit instances.
• Route indexed primary-key splits through the physical-position read path, recheck legacy raw-read eligibility, intersect row ranges with file-index pruning, subtract deletion vectors, and retain the complete original predicate for residual filtering.
• Add PkSortedIndexFile::Build for writing one BTree payload from value-sorted source-group input.
• Add Java-generated Parquet and ORC fixtures covering historical snapshots, mixed indexed and unindexed files, multi-file source groups, deletion vectors, partitions, buckets, fallback branches, and disabled-index behavior.

The planner falls back conservatively when metadata or coverage is incomplete, BTree metadata is malformed, any source file has an unknown or nonzero delete-row count, index evaluation fails, positions are invalid, or a file would require more than 4096 ranges. Snapshot-loading and index-evaluation failures are logged before fallback. Indexed splits retain their deletion-file association, and externally supplied indexed splits are rechecked before raw dispatch, so using the index does not change row visibility or query results.

The optimization applies to snapshot-scoped, non-read-optimized primary-key batch scans outside the Data Evolution path. Source splits that are not eligible for file-local reads remain unchanged; generated IndexedSplit instances are routed explicitly through the physical-position reader.

Only the BTree payload reader is wired into scan planning. Bitmap, vector, and full-text definitions are recognized but fall back to ordinary scans. Automatic payload construction and maintenance during compaction remain outside this change.

Tests

• Built paimon-common-test and paimon-core-test with GCC 8.3.0, C++17, Debug mode, ORC enabled, and -Wall -Werror.
• Ran the complete paimon-common-test suite: 1,441/1,441 passed.
• Ran the complete paimon-core-test suite: 1,728/1,728 passed.
• Ran paimon-primary-key-sorted-index-inte-test: 12/12 passed across Parquet and ORC.
• Ran the Apache Paimon Java 2.0 fixture generator test: 1/1 passed.
• Ran ci/scripts/test_cmake_modules.sh; all AArch64 -march and target-architecture checks passed.
• Ran pre-commit run --all-files.
• Ran check-clang-tidy on all changed C++ files.
• Ran Apache RAT over all 26 newly added non-fixture files: 26/26 Apache-licensed.
• Ran git diff --check.

The integration and regression tests cover snapshots 2/3/5, mixed indexed and unindexed files, multi-source payload localization, deletion vectors with residual predicates, legacy unknown/nonzero delete counts, malformed BTree metadata, safe AND/OR fallback, disabled indexes, partitions and buckets, fallback-branch routing, and unsupported scored splits.

API and Format

• Adds public option constants for pk-btree.index.columns, pk-bitmap.index.columns, pk-vector.index.columns, and pk-full-text.index.columns.
• Keeps the existing GlobalIndexer::CreateReader API. BTree readers create their private four-thread executor only when multiple payloads can submit work; single-payload reads, including primary-key groups, stay inline.
• Introduces no new storage format or protocol. It consumes the existing GlobalIndexMeta._SOURCE_META carrier and source-metadata v1 layout used by Java 2.0.
• File-name encoding is compatible with Java writeUTF for ASCII and non-NUL BMP UTF-8 names. Complete modified UTF-8 support for NUL and supplementary code points is not included.

Documentation

Adds docs/source/user_guide/primary_key_global_index.rst and links it from the user guide index. The page documents prerequisites, validation rules, fallback behavior, compatibility, and current limitations.

Generative AI tooling

Generated-by: OpenAI Codex (GPT-5) and Claude Code (Fable 5)

[wangyong9999]

cc @lxy-9602 @lucasfang could you please take a look when convenient? Thanks~

[lxy-9602]

@lszskye Please evaluate whether using GlobalDefaultExecutor here could cause any issues. I’d lean toward passing the same executor to each reader instead.

Also, the executor thread count in Java is based on GLOBAL_INDEX_THREAD_NUM, rather than using the machine core count as GetGlobalDefaultExecutor() does.

A more complete approach would be to let C++ GlobalIndexer: accept an external executor and, like Java, have the scan layer create and pass it in based on global-index.thread-num, instead of having BTreeGlobalIndexer implicitly choose GetGlobalDefaultExecutor().

[wangyong9999]

Updated. GlobalIndexer::CreateReader now accepts an external executor. PrimaryKeyIndexBatchScan creates one scan-scoped executor from global-index.thread-num and shares it across BTree readers; when unset, it preserves the existing C++ CPU-count default. The legacy GlobalIndexScanImpl path keeps its outer executor separate to avoid nested synchronous work on the same fixed-size pool. No executor is created when the plan has no valid index group.

[wangyong9999]

Follow-up correction: each primary-key index group currently contains exactly one payload, and UnionGlobalIndexReader evaluates a single reader inline, so the scan-scoped executor could never receive work. It has been removed, and the PK path now explicitly passes nullptr. The four-argument BTree reader retains its pre-PR private four-thread executor behavior for GlobalIndexScanImpl; global-index.thread-num is not applied to the single-payload PK path.

[lxy-9602]

This seems to be for binary compatibility of the source data file name in PrimaryKeyIndexSourceMeta, but in Java, under what scenarios would the file name contain Chinese characters or other special characters?

[lxy-9602]

I think I have a rough understanding of the issue now. It’s not limited to this spot—FileIndexFormat, DataSplit.bucketPath, and DeletionFile.path all have similar problems with Chinese characters when converting string to UTF.

I’d suggest removing this part from the current PR for now, and then submitting a separate PR later to fully address all writeUTF / readUTF related issues.

[wangyong9999]

Configured data-file.prefix can introduce Unicode. However, this is a stream-wide issue, so the PR-local codec has been removed and source metadata now uses the existing stream primitives. A separate change can address writeUTF / readUTF consistently across FileIndexFormat, DataSplit.bucketPath, DeletionFile.path, and source metadata. The remaining supplementary-code-point limitation is documented.

[lxy-9602]

Could we move the Family family parameter before options?

[wangyong9999]

Updated. Family family now precedes options; the member and accessor order and all call sites were changed consistently.

[lxy-9602]

I’m a bit curious whether ValidateNoDuplicates and ValidateUniqueColumns could be refactored into a shared helper function, with different error reporting as needed. Also, could we move the output parameter to the end of the parameter list?

[wangyong9999]

Updated. Both validation paths now reuse AddUniqueColumns with a caller-supplied duplicate-error callback, and the output set parameter is last.

[lxy-9602]

Please use ASSERT_OK_AND_ASSIGN instead of calling PAIMON_RETURN_NOT_OK first and then accessing value(). If there are similar cases elsewhere, could you fix them as well?

[wangyong9999]

Updated. This is production code, so it now uses PAIMON_ASSIGN_OR_RAISE rather than the test-only ASSERT_OK_AND_ASSIGN. Similar patterns in the new tests were changed to ASSERT_OK_AND_ASSIGN or EXPECT_OK_AND_ASSIGN.

[lxy-9602]

Please make the error message explicit. I’d recommend using ASSERT_NOK_WITH_MSG.

[wangyong9999]

Updated. Both duplicate-column tests now use ASSERT_NOK_WITH_MSG and check the exact diagnostics.

[lxy-9602]

Please try to reuse DataOutputStream, DataInputStream, MemorySegmentOutputStream for endianness conversion and data input/output.

[wangyong9999]

Updated. Deserialization now uses ByteArrayInputStream with DataInputStream, and serialization uses MemorySegmentOutputStream for pooled big-endian output. DataOutputStream cannot wrap MemorySegmentOutputStream without an adapter or extra copy. The string length and bytes remain explicit because MemorySegmentOutputStream::WriteString narrows through int16_t, while this wire field supports the full uint16_t range. Boundary and null-input tests were added.

[lxy-9602]

Please change this to something like:

if (is_external_path) {
    PAIMON_ASSIGN_OR_RAISE(Path path, PathUtil::ToPath(io_meta.file_path));
    external_path = path.ToString();
}

Could we normalize the path here? Regular global index writing does this in global_index_write_task.cpp. In Java, org.apache.paimon.fs.Path parses and normalizes the URI during construction, so the final stored value is the normalized form from Path.toString().

[wangyong9999]

Updated. External payload paths are now normalized through PathUtil::ToPath and Path::ToString, with regression coverage for duplicate separators.

[lxy-9602]

The current implementation puts all planned sorted data for an entire index file into a single sorted_values, which could lead to excessive memory usage. While Java uses an external sort buffer here.

Given the scope of the current PR, I’d suggest adding a TODO to clearly document this as a known issue and plan to fix it in a follow-up PR.

[wangyong9999]

Added an author-tagged TODO to replace the in-memory input with an external sort buffer and bounded writer batches.

[lxy-9602]

Please avoid using default arguments in production code.

[lxy-9602]

Also, I’m not sure it’s necessary to add a new CreateReader function. Could we just extend the existing one with a local_row_ranges parameter instead?

[wangyong9999]

Updated. The existing overloads now take local_row_ranges explicitly; the forwarding overload and production default argument were removed, and callers pass std::nullopt explicitly.

[lxy-9602]

If the indexed split contains scores, is read supported right now? I don’t see any handling logic for that. If it’s not supported, please fail fast and add a TODO to mark it clearly.

[wangyong9999]

Updated. Primary-key reads now fail fast for scored IndexedSplits before routing or force-keep-delete fallback; a TODO and regression coverage were added.

[lxy-9602]

It seems in Java, rawConvertible is always false here. The current implementation would make the scan results inconsistent with Java.

[wangyong9999]

Updated. Derived single-file splits now use rawConvertible(false), and IndexedSplits are routed explicitly to the physical-position reader while unindexed splits use the normal merge path.

[lxy-9602]

PkSortedIndexGroup::Create returns optional<PkSortedIndexGroup>, but the call sites seem to use std::shared_ptr<PkSortedIndexGroup>. Should we adjust the return type of Create for consistency?

[wangyong9999]

Updated. Create now returns std::shared_ptr, and bucket state and file plans reuse the same group instance.

[lxy-9602]

Could you help clarify the distinction between QueryOperation and Function? They seem somewhat overlapping to me, so I’m wondering why both are needed separately.

[wangyong9999]

Agreed. QueryOperation was only a cache-key tag, so it has been removed and the cache now uses Function::Type directly.

[lxy-9602]

Could ScalarDefinitions be handling something similar here?

[wangyong9999]

Updated. Scalar-family filtering is centralized in PrimaryKeyIndexDefinitions::ScalarDefinitions and reused by planning and evaluation.

[lxy-9602]

ASSERT_NOK_WITH_MSG(Func(), error_message); is OK here.

[wangyong9999]

Updated. The Result is now passed directly to ASSERT_NOK_WITH_MSG.

[lxy-9602]

ASSERT_NOK(PrimaryKeySortedIndexScan::CreatePlan(
kSnapshotId, {split}, definitions_, MakeEntries(payload)));

[wangyong9999]

Updated. ASSERT_NOK now checks CreatePlan directly.

[lxy-9602]

The newly added tests mainly cover internal logic such as CreatePlan, Evaluate, and ToSplits. Please also add E2E tests similar to test/inte/global_index_test.cpp, covering the complete path:

Java writes the table and PK BTree index
    → C++ TableScan::CreatePlan
    → IndexedSplit / DataSplit
    → TableRead::CreateReader
    → validate the returned rows

These tests should not only invoke PkSortedIndexFile::Build, PrimaryKeySortedIndexScan::Evaluate, or mock readers directly. Otherwise, they do not cover Java/C++ format compatibility, manifest/source metadata parsing, index reader creation, split conversion, and actual data-file reading.

Please consider covering the following scenarios:

1. Index written by Java and read by C++

   Let Java write a primary-key table with a pk-btree index, then query it from C++ using a predicate on the indexed field.

   In addition to validating the returned rows, verify that the scan plan contains an IndexedSplit with the expected single data file and file-local row ranges.

2. A mixture of indexed and non-indexed files

   First compact the table to generate indexed files, then append new L0 files without another compaction. The query range should match rows from both:

   • a compacted file covered by a PK BTree index;
   • a newly appended L0 file without an index.

   Verify that the plan contains both IndexedSplit and regular DataSplit instances, and that the final read returns rows from both groups without false negatives.

3. One index payload covering multiple source data files

   Use Java to perform a compaction that produces multiple data files while creating only one PK BTree index group. The query result should span at least two source files.

   Verify that C++ correctly converts group-global row ordinals into file-local row ranges and creates the corresponding single-file IndexedSplit instances. Execute TableRead and verify that matching rows from all source files are returned.

4. Deletion Vector together with a residual predicate

   Let Java compact the table and build the index, then perform updates or deletes to generate a deletion vector. Use a predicate containing both an indexed and a non-indexed field, for example:

   score = 10 AND tag = 'keep'
   

   Verify that the PK BTree row ranges, deletion vector, and complete residual predicate are all applied. Deleted rows and rows that fail the residual predicate must not be returned.

5. Safe fallback for AND/OR predicates

   Test both:

   score = 10 AND tag = 'keep'
   score = 10 OR tag = 'keep'
   

   For the AND predicate, the score index may narrow the row ranges while tag remains a residual predicate.

   For the OR predicate, if one branch cannot be evaluated by the index, the scan must not prune data using only the indexed branch. It should safely fall back to a regular scan. Both cases should execute the complete scan-and-read path and validate the final rows.

6. Global index disabled

   Set the scan option:

   {{Options::GLOBAL_INDEX_ENABLED, "false"}}

   or equivalently:

   global-index.enabled=false
   

   Verify that no IndexedSplit is produced, the regular scan/read path is used, and the returned rows are identical to those returned with the index enabled.

7. Historical snapshots and index versions

   Read the following snapshots from the same Java-generated fixture:

   • snapshot 2: use the index generated by the first compaction;
   • snapshot 3: use the old index for the compacted file and fall back for the new L0 file;
   • snapshot 5: use the index rebuilt by the later compaction.

   Validate both the plan and the final rows for each snapshot. This should ensure that C++ only applies index payloads belonging to the requested snapshot and matching the exact source files, rather than incorrectly applying the snapshot 5 index to files from snapshot 2 or 3.

8. Partitions and multiple buckets

   Prepare a Java fixture with multiple partitions and multiple buckets, with PK BTree index groups in different buckets. The query should match rows across partitions and buckets.

   Verify that each IndexedSplit has the correct partition, bucket, source data file, and index-file path, and validate the final read result.

9. Routing IndexedSplit through FallbackTableRead

   Configure scan.fallback-branch and create a scenario where the main branch produces an IndexedSplit.

   Verify that:

   • an IndexedSplit from the main branch is sent to the main table reader;
   • a split from the fallback branch is sent to the fallback table reader;
   • an IndexedSplit is not rejected or routed to the wrong reader because of its split type;
   • the final read result is correct.

10. Both Parquet and ORC

    Please generate equivalent Java fixtures for Parquet and ORC and parameterize the tests with TEST_P, for example:

    using ParamType = std::string;
    
    INSTANTIATE_TEST_SUITE_P(
        FileFormat,
        PrimaryKeySortedIndexE2ETest,
        ::testing::Values("parquet", "orc"));

    Every case should cover the actual TableScan + TableRead path rather than only validating the plan. Depending on the scenario, the tests should also verify the split type, source data file, file-local row ranges, and deletion file.

These tests could be added to a dedicated primary_key_sorted_index_inte_test.cpp or to the existing global_index_test.cpp. The Java-generated fixtures should be stored under test/test_data/{parquet,orc}/..., with a README documenting the schema, writes, compactions, and expected state of each snapshot.

[wangyong9999]

Added Java-generated Parquet/ORC fixtures and parameterized TableScan-to-TableRead E2E coverage for indexed plus visible unindexed APPEND-source files, multi-file payloads and file-local ranges, deletion vectors with AND residual filtering, safe OR fallback, disabled indexes, snapshots 2/3/5, partitions and buckets, and fallback-branch routing. The fixture READMEs document the exact snapshot construction.

[lxy-9602]

Since pk is currently always passed as nullptr, is it really necessary to add this extra public interface with the executor?

[wangyong9999]

Good point. PK groups contain one payload, so the executor overload was unnecessary. I removed it and kept the existing four-argument API; BTree now creates its private executor only for multiple payloads, preserving legacy parallel evaluation while single-payload PK reads stay inline.

[lxy-9602]

+1

[SteNicholas]

@wangyong9999, I have found two correctness issues in the source-backed primary-key BTree index read path. PTAL.

[SteNicholas]

LGTM.