[feature](single replica ingest) support single replica ingest binlog

[Ryan19929]

What problem does this PR solve?

Issue Number: close #xxx

Related PR: HYDCP/hy-ccr-sycner#3

Problem Summary:
Currently, CCR cross-cluster binlog ingest downloads the same rowset files on every replica, causing redundant cross-cluster traffic and slower ingest. This PR introduces single-replica ingest binlog: the leader BE downloads the rowset once and distributes the files to follower BEs, reducing traffic and improving performance.

Release note

Add feature flag feature_single_replica_ingest_binlog (default disabled). When enabled, CCR ingests binlog through a single leader replica and distributes rowset files to followers locally.

Check List (For Author)

• Test

  • Regression test
  • Unit Test
  • Manual test (add detailed scripts or steps below)
  • No need to test or manual test. Explain why:
    • This is a refactor/code format and no logic has been changed.
    • Previous test can cover this change.
    • No code files have been changed.
    • Other reason

• Behavior changed:

  • No.
  • Yes.
    New BE-side single-replica ingest path and follower distribution. The legacy multi-replica path remains the default and is used when the feature flag is off.

• Does this need documentation?

  • No.
  • Yes.
    Document PR is pending.

Check List (For Reviewer who merge this PR)

• Confirm the release note
• Confirm test cases
• Confirm document
• Add branch pick label

[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?

[wenzhenghu]

Reviewed the current head 4af7bc3f4fc421beb2c72c8eb6ee075d2d0d3d37. I found one P1 and four P2 issues that should be addressed before approval.

1. [P1] Propagate delete-bitmap errors before committing the ingested rowset (be/src/service/backend_service.cpp:155-160)

   commit_ingested_rowset() discards the statuses returned by both BaseTablet::commit_phase_update_delete_bitmap() and CalcDeleteBitmapToken::wait(). If task submission fails synchronously, or an asynchronous delete-bitmap task records an error, the function still executes commit_txn() and then attaches an incomplete delete bitmap. The new single-replica leader and follower paths both call this helper, so a Unique Key MoW rowset can be committed with incorrect visibility information.

   Please preserve the error-propagation pattern from [fix](be) Propagate CCR delete bitmap calculation errors #67143 when resolving the overlap: save the submission status, always call wait() to drain already-submitted tasks, use the wait status when submission succeeded, and return kError before commit_txn() on either failure. This should also have focused tests for synchronous submission failure and asynchronous calculation failure.

2. [P2] Avoid eagerly creating hardware_concurrency permanent threads while the feature is disabled (be/src/service/backend_service.cpp:1283-1302)

   The follower-distribution pool is created unconditionally at BE startup. With the default value, min_threads == max_threads == hardware_concurrency; ThreadPool::init() immediately creates all minimum threads. Therefore, even though feature_single_replica_ingest_binlog is disabled by default in CCR Syncer, every BE can gain dozens or hundreds of permanent threads.

   Please use min_threads=0 with the configured value as max_threads, or lazily construct the pool on the first single-replica request.

3. [P2] Increment the cleanup success metric only after deletion succeeds (be/src/service/backend_service.cpp:460, :843)

   binlog_ingest_redundant_rowset_cleanup_success_total is incremented before _delete_downloaded_files() calls batch_delete(). When deletion fails, both the success and failed counters increase. This produces misleading operational data and allows a regression check to report success while redundant files remain.

   Please return the cleanup Status, or move both success/failed metric updates into the cleanup helper so they reflect the actual outcome.

4. [P2] Fix the nonexistent metric name in the retry regression (regression-test/suites/ccr_syncer_p0/test_single_replica_ingest_binlog.groovy:354)

   The test reads doris_be_binlog_ingest_redundant_files_deleted_total, but this PR registers doris_be_binlog_ingest_redundant_rowset_cleanup_success_total. On this head, readMetric() returns -1 before and after the retry, so the subsequent after > before assertions cannot pass. After correcting the name, please also assert that the metric was found instead of accepting -1 as a baseline.

5. [P2] Run the required clang-format 16 formatter

   An exact-head clang-format 16 dry run fails for:

   • be/src/service/backend_service.cpp
   • be/src/service/backend_service_ingest_helper.h
   • be/src/storage/txn/txn_manager.cpp
   • be/test/service/backend_service_ingest_test.cpp

   Please run ./build-support/clang-format.sh and recheck the resulting diff.

Additional review conclusions:

• The optional Thrift fields preserve wire compatibility. The Syncer-side missing-success-list and follower-failure handling provides a reasonable old/mixed-BE fallback.
• The future/reference lifetimes in follower fan-out are safe because all futures are drained before the function returns; I did not find a new lock-order or deadlock issue.
• Leader-first commit followed by follower retry/fallback is coherent with FE commit-info collection, and per-replica local rowset IDs are expected.
• The BE unit test covers kAlreadyExist without overwriting the existing MoW delete bitmap, but it does not cover the ignored delete-bitmap error paths above.

Validation boundary: the real PR patch is the 14-file diff from merge-base 9673a5003666b858f755188ce293c7ab5cc3d1c5 to the reviewed head (+1879/-128). git diff --check passed. I did not independently run a BE build, BE unit tests, or the two-cluster CCR regression. The PR is currently Draft and conflicts with current master; it should be reviewed again after conflict resolution because the resulting production diff may change materially.

[Ryan19929]

已按 review 意见修复 4 个 P2 问题并 rebase 到最新 master，当前 head 6be3fa1。

• P2-1：follower distribution pool 改为 set_min_threads(0)，默认不再在 BE 启动时预创建 permanent 线程。
• P2-2：_delete_downloaded_files 改为返回 Status，success/failed 两个 cleanup 指标都在 helper 内根据 batch_delete 实际结果更新。
• P2-3：回归测试中指标名改为 doris_be_binlog_ingest_redundant_rowset_cleanup_success_total，并增加 metric 必须被找到（返回值 != -1）的断言。
• P2-4：使用 clang-format 16 格式化 review 列出的 4 个文件，本地 ./build-support/check-format.sh 已通过，git diff --check 通过。

P1（delete-bitmap error propagation）已按说明忽略，等待 #67143。请再次 review。