[core] Wire sorted primary-key indexes into writes

[JingsongLi]

Purpose

Wire source-backed BTree and Bitmap indexes into the primary-key table write lifecycle.

Changes

• Create a unified bucket-local coordinator for vector, BTree, and Bitmap primary-key index definitions.
• Restore source-backed payloads and construct sorted-index builders from the table schema, bucket path, reader factory, and shared IOManager.
• Filter restored payloads by indexType and field ID at each maintainer boundary, so scalar payloads never enter vector state.
• Stage all definition results and publish caller increments only after every definition succeeds; abort prepared vector/scalar commits in reverse order and clean generated payloads when a later definition fails or is interrupted.
• Run multiple scalar index definitions sequentially while allowing a failed definition to remain uncovered without suppressing other definitions.
• Publish index additions and deletions through the append or compact increment that owns the corresponding source-file transition.
• Enable the file-store writer whenever BTree or Bitmap primary-key index columns are configured.
• Keep source-backed payloads out of the ordinary global row-id scanner and coverage calculation, so both index models can coexist safely.
• Keep Postpone Bucket writes unindexed in bucket -2; the fixed-bucket runtime view used by background compaction creates and publishes the coordinator.
• Cover mixed Vector+BTree+Bitmap write/restore, coordinator interruption rollback/retry, and non-blocking scalar scheduling without the prior latch race.

Impact

Compacted primary-key table files now automatically receive configured BTree and Bitmap payloads. Read-side source pruning remains intentionally deferred to a follow-up PR.

Validation

• mvn -pl paimon-codegen-loader -am -Pfast-build -DskipTests package
• mvn -pl paimon-core -am -Pfast-build -DfailIfNoTests=false -DwildcardSuites=none -Dtest=BucketedVectorIndexMaintainerTest,PkVectorBucketIndexStateTest,BucketedSortedIndexMaintainerTest,BucketedPrimaryKeyIndexMaintainerTest,PrimaryKeySortedIndexMaintenanceTest,PrimaryKeyIndexWriteTest,PrimaryKeyVectorIndexWriteTest,BtreeGlobalIndexTableTest test (46 tests)
• mvn -pl paimon-core -DskipTests validate

[leaves12138]

I found two blocking correctness issues in the mixed/multi-index write lifecycle, plus a reproducible failure in the new concurrency test.

[leaves12138]

restoredPayloads is the full result of scanSourceIndexes, so it contains the payloads for every configured source-backed definition. Passing that unfiltered list into the vector factory makes PkVectorBucketIndexState validate BTree/Bitmap payloads as vector segments and throw Vector payload ... has a different index type. A table configured with both a vector index and a scalar index therefore cannot restore a bucket writer after scalar payloads have been committed. Please filter the restored payloads by the vector field/index type before creating the vector maintainer, and add a mixed vector + scalar write/restore test.

[leaves12138]

The coordinator publishes each maintainer's state and merges its files into the caller increments before later maintainers have succeeded, but there is no coordinator-level rollback. If an earlier vector/scalar definition completes and a later blocking scalar prepareCommit is interrupted, the overall call throws while the earlier definition remains covered and its generated payload remains in the increment. Retrying or discarding that failed prepare can then omit the payload or leave it orphaned. Individual maintainers only roll back their own invocation. Please make this operation atomic across all definitions (or defer publication until all definitions succeed) and cover an interruption after the first of two scalar definitions completes.

[leaves12138]

This test is racy and fails consistently on my machine. firstCompleted is counted down inside the callable immediately before it returns, so the following non-blocking prepareCommit may still observe Future.isDone() == false and return without starting the second definition; the await here then times out. Please wait/poll until the first future is actually consumed (for example, retry non-blocking prepare until the second build starts) instead of treating the in-callable latch as future completion.

[leaves12138]

Re-reviewed the updated head. The mixed vector/scalar restore filtering, coordinator-wide rollback, and scalar build concurrency regression are addressed. The focused core test suite and repeated concurrency test pass locally.