support async publish for cloud mow

[bobhan1]

Motivation

In cloud-native (disaggregated storage-compute) mode, high-frequency concurrent loads on MOW (Merge-on-Write) tables suffer from a serious throughput bottleneck that cannot be alleviated by horizontally scaling resources.

How It Works in Shared-Nothing Mode (Baseline)

In shared-nothing mode, MOW loads use a two-phase commit:

• Each load reserves its version during the commit phase — subsequent loads can immediately grab the next version without waiting for the current one to publish.
• Delete bitmap computation tasks are distributed across BE nodes, with tablet metadata stored in each BE's local RocksDB.
• Conflicts between loads and compactions are resolved at the tablet level via in-memory locks on each BE.

This architecture scales horizontally: adding more BEs reduces commit + publish latency and increases throughput.

The Problem in Cloud Mode

Cloud mode currently uses a single-phase commit. The entire process runs serially under an FE table-level in-memory write lock:

1. Acquire an MS table-level distributed lock (competing with all compactions on that table)
2. Fetch partition versions from MS
3. Dispatch delete bitmap computation tasks to BEs and synchronously wait for completion
4. Submit the transaction to MS (read/write KVs proportional to the number of tablets involved)

This causes:

• Complete serialization of all concurrent loads — scaling BEs only speeds up delete bitmap computation but cannot remove the serial bottleneck
• High tail latency — loads touching many tablets cause slow MS commits, and concurrent lock contention makes latency unpredictable
• FDB pressure — large transactions with many tablet KV operations can slow down the entire MS

Solution

Introduce a two-phase async publish mode for cloud MOW tables, controlled by the table property enable_mow_async_publish (new tables only). The core idea is to move the expensive publish operations (delete bitmap calculation + rowset conversion) out of the lock-holding path and execute them asynchronously.

Phase 1: Commit (Fast, Brief FE Table Lock)

1. FE acquires the table-level in-memory write lock
2. Calls MS commit_txn with enable_mow_async_publish=true:
   • Reserves a commit version for each partition (new partition_commit_version_key, separate from visible version)
   • Persists transaction metadata into TxnInfoPB (involved tablets, commit versions, load schema params, etc.)
   • Transitions transaction status to COMMITTED
3. Adds the transaction to FE's in-memory CommittedTxnManager
4. Releases the table lock — lock hold time is minimal, the next load can enter commit immediately

Phase 2: Publish (Async, No FE Table Lock)

Driven by the new CloudPublishDaemon background thread:

1. Dispatch delete bitmap tasks: Groups tablets by BE and dispatches CalcDeleteBitmapTasks. Failed tasks auto-retry via AgentTaskQueue / BE heartbeat mechanism.
2. BE-side processing:
   • Acquire MS tablet-level lock (replaces the old table-level lock)
   • Compute incremental delete bitmap
   • Call MS convert_tmp_rowset to convert tmp rowset to formal versioned rowset (per-tablet, independent)
   • Best-effort local apply (add rowset + delete bitmap to tablet memory to reduce subsequent query sync overhead)
3. Lightweight publish: Once all BE tasks complete, call MS commit_txn with is_lightweight_publish=true:
   • Only advance partition visible version (commit version → visible version)
   • Update transaction status to VISIBLE
   • No tmp rowset or pending delete bitmap handling needed

Key Design Decisions

Aspect	Old (Single-Phase)	New (Two-Phase Async Publish)
Lock granularity	MS table-level lock	MS tablet-level lock
Version management	Single partition version	Separate commit version and visible version
Pending delete bitmap	Must clean up previous failure residue	Eliminated (committed txns must succeed)
Tmp rowset conversion	Batched in final commit_txn	Per-tablet independent convert_tmp_rowset
FE table lock hold time	Entire commit + publish	Commit phase only
Fault tolerance	Any failure before MS commit → txn fails	After commit → must succeed; FE restart recovers from MS

Implementation Details

Proto & KV Schema (gensrc/proto/cloud.proto, cloud/src/meta-store/keys.cpp)

• New TxnTabletInfoPB message: records per-tablet BE endpoint info for publish-phase task dispatch
• TxnInfoPB new fields: committed_versions, involved_tablets, load_schema_param, published_tablet_ids
• CommitTxnRequest new routing flags: enable_mow_async_publish and is_lightweight_publish
• New convert_tmp_rowset RPC with ConvertTmpRowsetRequest / ConvertTmpRowsetResponse
• New KV keys: partition_commit_version_key (commit version storage), meta_delete_bitmap_tablet_lock_key (tablet-level lock)

Meta Service (cloud/src/meta-service/)

• commit_txn_async_publish: Commit-phase handler — reserves commit versions, persists txn metadata; idempotent (re-returns stored versions if already committed)
• commit_txn_2pc_lightweight_publish: Publish-phase handler — advances visible versions with strict version continuity validation
• convert_tmp_rowset: Per-tablet tmp→formal rowset conversion; idempotent (detects duplicate by rowset_id at target version)
• Tablet-level lock: get_delete_bitmap_tablet_lock / remove_delete_bitmap_tablet_lock — fine-grained mutual exclusion between loads and compactions
• Skip pending delete bitmap processing for async publish tables in update_delete_bitmap

FE (fe/fe-core/.../cloud/transaction/)

• CloudGlobalTransactionMgr: New commitAndPublishTwoPhase flow — fast commit then awaitPublish(timeout) for async publish completion
• CloudPublishDaemon (new): MasterDaemon background thread with two-phase loop — dispatch calc tasks → try finish transactions via lightweight publish
• CommittedTxnManager (new): Thread-safe committed transaction tracking with dual indexes (txnId + tableId)
• CommittedTxnEntry (new): Per-transaction state tracking + CountDownLatch-based publish await mechanism
• TxnUtil (new): Shared utilities (commit callbacks, retry backoff)
• Table property enable_mow_async_publish support (OlapTable, TableProperty, PropertyAnalyzer)

BE (be/src/cloud/)

• CloudEngineCalcDeleteBitmapTask: Async publish path adds MS tablet-level lock acquisition → delete bitmap calculation → convert_tmp_rowset call → best-effort local apply (merge delete bitmap + add rowset to tablet memory)
• Compaction paths adapted for tablet-level lock

Tests

• cloud/test/txn_async_publish_test.cpp: 1500+ line MS integration test covering commit/publish/convert_tmp_rowset normal and error paths
• CloudPublishDaemonTest.java: FE publish daemon unit tests
• CloudGlobalTransactionMgrTest.java, OlapTableTest.java, CreateTableTest.java: Related unit tests

Risks & Limitations

• Publish phase has a strong dependency on FE (FE failover/restart requires recovering committed txn info from MS)
• Additional FE memory usage for committed transaction state
• Currently limited to single-table transactions
• Interaction with sub-transactions and snapshot chains not considered yet

Test Plan

• MS-side txn_async_publish_test passes
• FE-side CloudPublishDaemonTest passes
• Table property creation and parsing tests pass
• End-to-end: high-concurrency stream load on single MOW table
• FE failover recovery test
• Concurrent load + compaction stress test
• Performance comparison: large tablet count load throughput (old vs new)

[hello-stephen]

Thank you for your contribution to Apache Doris.
Don't know what should be done next? See How to process your PR.

Please clearly describe your PR:

1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?