[test](be) Repeat and document async cache write benchmark

### What problem does this PR solve?

Issue Number: None

Related PR: #65658

Problem Summary: The asynchronous file-cache write microbenchmark previously emitted only one sample per case and did not establish the storage baseline of the cache filesystem. These short concurrent cases are sensitive to scheduler activity, page-cache state, filesystem metadata, and background writeback, so a single number can hide material variance and make worker-scaling conclusions unreliable.

Run every selected reader, service, and inflight-index case five times by default and add the one-based repetition to each machine-readable RESULT line. Add an installed runner that automatically detects fio, measures direct 1 MiB sequential QD1 and random QD16 writes on the same filesystem, then starts the benchmark. The runner uses a unique sibling directory under the selected cache path, unlinks fio data, and keeps direct I/O out of the page cache. The benchmark now rejects non-empty cache paths instead of recursively clearing them, and suppresses INFO logging so merged stdout and stderr cannot corrupt RESULT records.

Expand the tool README with the component flow, coverage and non-goals of each group, default workload, field semantics, fio controls, cache-path ownership, repetition methodology, and interpretation guidance. Median is the primary value and the observed min-to-max range is retained.

Release experiment on /dev/nvme11n1 ext4, 1 MiB blocks, 64 KiB caller reads, 16 producers, 128 reader operations, 256 service attempts, and five repetitions:

- fio direct baseline: sequential QD1 was 2513 MiB/s with 161 us p95 completion latency; random QD16 was 3106 MiB/s with 10.552 ms p95 completion latency.
- Reader foreground throughput in ops/s, median [min, max]: sync 5645 [5124, 7649], async 6739 [4739, 8298]. Median average latency was 1459 us for sync and 912 us for async. The median throughput improved 19.4% and median average latency fell 37.5%, while the overlapping ranges show why repeated samples are required.
- Verified service completion in MiB/s, median [min, max]: 1 worker 798 [730, 968], 4 workers 1562 [1260, 1751], and 16 workers 7825 [5255, 13203]. Median drain time fell from 0.300 s to 0.153 s and 0.014 s. These are buffered append/finalize completions without fsync and are not durable-media throughput.
- Backpressure accepted 76 [64, 101] and rejected 180 [155, 192] tasks, with peak pending fixed at the configured limit of 64, peak queued fixed at 48, and peak inflight 65 [65, 67]. Every accepted task was verified as persisted.
- Inflight lookup throughput in ops/s, median [min, max]: sharded miss 5.531M [4.028M, 6.621M], sharded hit 4.198M [3.270M, 6.036M], and hot-key hit 1.104M [1.090M, 1.268M].
- All 45 RESULT records were complete and parseable. All reader ranges and all accepted service tasks passed final BlockFileCache coverage verification.

### Release note

None

### Check List (For Author)

- Test: Manual test
    - ./build.sh --be --file-cache-microbench -j100 (Release)
    - ./output/be/bin/run-async-file-cache-write-microbench.sh --benchmark_mode=all --cache_path=./output/async_file_cache_write_microbench_repeat_5_clean --producer_threads=16 --reader_workers=16 --worker_counts=1,4,16 --repetitions=5
    - Non-empty cache-path rejection with sentinel preservation
    - build-support/clang-format.sh
    - build-support/check-format.sh
    - bash -n and shellcheck for the runner
    - git diff --check
- Behavior changed: No (benchmark tooling only)
- Does this need documentation: No (tool README updated)

Author: bobhan1

be/src/io/tools/async_file_cache_write_microbench.cpp

@@ -81,6 +81,7 @@ DEFINE_int32(producer_threads, 16, "Concurrent foreground readers or task produc
 DEFINE_int32(reader_workers, 16, "Async write workers used by the reader comparison");
 DEFINE_string(worker_counts, "1,4,16",
               "Comma-separated async write worker counts used by service scaling cases");
+DEFINE_uint64(repetitions, 5, "Measured repetitions of every selected benchmark case");
 DEFINE_uint64(backpressure_pending_tasks, 64,
               "Pending-task limit used by the saturated service case");
 DEFINE_uint64(queue_sample_interval_us, 50,
@@ -215,7 +216,8 @@ Status validate_flags(const std::vector<std::string>& modes,
     if (FLAGS_service_task_size == 0 || FLAGS_service_task_size > FLAGS_block_size ||
         FLAGS_service_key_count == 0 || FLAGS_index_operations_per_thread == 0 ||
         FLAGS_index_key_count == 0 || FLAGS_backpressure_pending_tasks == 0 ||
-        FLAGS_queue_sample_interval_us == 0 || FLAGS_timeout_seconds == 0) {
+        FLAGS_queue_sample_interval_us == 0 || FLAGS_timeout_seconds == 0 ||
+        FLAGS_repetitions == 0) {
         return Status::InvalidArgument(
                 "operation counts, timeouts, and 0 < service_task_size <= block_size are required");
     }
@@ -390,7 +392,7 @@ class BenchmarkEnvironment {
             ExecEnv::GetInstance()->set_file_cache_factory(nullptr);
             _factory.reset();
         }
-        if (!FLAGS_keep_cache) {
+        if (_owns_cache_path && !FLAGS_keep_cache) {
             std::error_code error;
             std::filesystem::remove_all(FLAGS_cache_path, error);
         }
@@ -399,6 +401,55 @@ class BenchmarkEnvironment {
     /// Configure globals, create the cache, and wait for its asynchronous metadata open.
     /// @param cache_capacity Bytes reserved for the benchmark's normal queue.
     Status initialize(size_t cache_capacity) {
+        std::error_code error;
+        const auto cache_path =
+                std::filesystem::absolute(FLAGS_cache_path, error).lexically_normal();
+        if (error) {
+            return Status::IOError("failed to resolve cache path {}: {}", FLAGS_cache_path,
+                                   error.message());
+        }
+        const auto current_path = std::filesystem::current_path(error);
+        if (error) {
+            return Status::IOError("failed to resolve current directory: {}", error.message());
+        }
+        if (cache_path == cache_path.root_path() || cache_path == current_path) {
+            return Status::InvalidArgument("cache path must be a dedicated subdirectory: {}",
+                                           cache_path.string());
+        }
+        const bool cache_path_exists = std::filesystem::exists(cache_path, error);
+        if (error) {
+            return Status::IOError("failed to inspect cache path {}: {}", cache_path.string(),
+                                   error.message());
+        }
+        if (cache_path_exists) {
+            const bool cache_path_is_directory = std::filesystem::is_directory(cache_path, error);
+            if (error) {
+                return Status::IOError("failed to inspect cache path {}: {}", cache_path.string(),
+                                       error.message());
+            }
+            if (!cache_path_is_directory) {
+                return Status::InvalidArgument("cache path is not a directory: {}",
+                                               cache_path.string());
+            }
+            const bool cache_path_is_empty = std::filesystem::is_empty(cache_path, error);
+            if (error) {
+                return Status::IOError("failed to inspect cache path {}: {}", cache_path.string(),
+                                       error.message());
+            }
+            if (!cache_path_is_empty) {
+                return Status::InvalidArgument(
+                        "cache path must not exist or must be an empty directory: {}",
+                        cache_path.string());
+            }
+        } else {
+            std::filesystem::create_directories(cache_path, error);
+            if (error) {
+                return Status::IOError("failed to create cache path {}: {}", cache_path.string(),
+                                       error.message());
+            }
+        }
+        _owns_cache_path = true;
+
         config::enable_async_file_cache_write = true;
         config::enable_async_file_cache_write_inflight_write_buffer_index = true;
         config::enable_read_cache_file_directly = false;
@@ -423,14 +474,6 @@ class BenchmarkEnvironment {
         _factory = std::make_unique<FileCacheFactory>();
         ExecEnv::GetInstance()->set_file_cache_factory(_factory.get());
 
-        std::error_code error;
-        std::filesystem::remove_all(FLAGS_cache_path, error);
-        std::filesystem::create_directories(FLAGS_cache_path, error);
-        if (error) {
-            return Status::IOError("failed to create cache path {}: {}", FLAGS_cache_path,
-                                   error.message());
-        }
-
         FileCacheSettings settings;
         settings.capacity = cache_capacity;
         settings.max_file_block_size = FLAGS_block_size;
@@ -492,6 +535,7 @@ class BenchmarkEnvironment {
 private:
     std::unique_ptr<FileCacheFactory> _factory;
     BlockFileCache* _cache {nullptr};
+    bool _owns_cache_path {false};
 };
 
 /// Verify that a complete range can be resolved from the final cache state.
@@ -538,7 +582,8 @@ struct AsyncWriteResult {
 
 /// Print one machine-readable line without hiding the foreground/drain distinction.
 /// @param result Completed benchmark result.
-void print_async_write_result(const AsyncWriteResult& result) {
+/// @param repetition One-based repetition index.
+void print_async_write_result(const AsyncWriteResult& result, size_t repetition) {
     const double foreground_ops_per_sec =
             static_cast<double>(result.operations) / result.foreground_seconds;
     const double persisted_mib_per_sec =
@@ -548,9 +593,10 @@ void print_async_write_result(const AsyncWriteResult& result) {
                     : 0;
     std::cout << std::fixed << std::setprecision(3) << "RESULT"
               << " benchmark=" << result.benchmark << " variant=" << result.variant
-              << " producers=" << result.producers << " workers=" << result.workers
-              << " operations=" << result.operations << " accepted=" << result.accepted
-              << " rejected=" << result.rejected << " persisted=" << result.persisted
+              << " repetition=" << repetition << " producers=" << result.producers
+              << " workers=" << result.workers << " operations=" << result.operations
+              << " accepted=" << result.accepted << " rejected=" << result.rejected
+              << " persisted=" << result.persisted
               << " bytes_per_operation=" << result.bytes_per_operation
               << " foreground_seconds=" << result.foreground_seconds
               << " drain_seconds=" << result.drain_seconds
@@ -568,8 +614,9 @@ void print_async_write_result(const AsyncWriteResult& result) {
 /// @param environment Shared real cache, cleared before the case.
 /// @param mode Explicit write policy applied to every CachedRemoteFileReader.
 /// @param variant Stable output label.
-Status run_reader_case(BenchmarkEnvironment* environment, CacheWriteMode mode,
-                       std::string variant) {
+/// @param repetition One-based repetition index included in output.
+Status run_reader_case(BenchmarkEnvironment* environment, CacheWriteMode mode, std::string variant,
+                       size_t repetition) {
     DORIS_CHECK(environment != nullptr);
     RETURN_IF_ERROR(environment->clear_cache());
     RETURN_IF_ERROR(environment->configure_service(
@@ -684,7 +731,7 @@ Status run_reader_case(BenchmarkEnvironment* environment, CacheWriteMode mode,
             .peak_inflight = sampler.peak_inflight(),
             .latency = summarize_latencies(latencies),
     };
-    print_async_write_result(result);
+    print_async_write_result(result, repetition);
     return Status::OK();
 }
 
@@ -700,8 +747,9 @@ struct ServiceTaskRecord {
 /// @param variant Stable output label.
 /// @param workers Active service consumers.
 /// @param max_pending Bounded pending-task limit.
+/// @param repetition One-based repetition index included in output.
 Status run_service_case(BenchmarkEnvironment* environment, std::string variant, size_t workers,
-                        size_t max_pending) {
+                        size_t max_pending, size_t repetition) {
     DORIS_CHECK(environment != nullptr);
     RETURN_IF_ERROR(environment->clear_cache());
     RETURN_IF_ERROR(environment->configure_service(workers, max_pending));
@@ -847,16 +895,18 @@ Status run_service_case(BenchmarkEnvironment* environment, std::string variant,
             .peak_inflight = sampler.peak_inflight(),
             .latency = summarize_latencies(latencies),
     };
-    print_async_write_result(result);
+    print_async_write_result(result, repetition);
     return Status::OK();
 }
 
 /// Print one inflight-index result using the same latency field names as write cases.
-void print_index_result(std::string_view variant, size_t operations, double elapsed_seconds,
-                        const LatencySummary& latency) {
+/// @param repetition One-based repetition index.
+void print_index_result(std::string_view variant, size_t repetition, size_t operations,
+                        double elapsed_seconds, const LatencySummary& latency) {
     std::cout << std::fixed << std::setprecision(3) << "RESULT benchmark=index"
-              << " variant=" << variant << " producers=" << FLAGS_producer_threads
-              << " operations=" << operations << " elapsed_seconds=" << elapsed_seconds
+              << " variant=" << variant << " repetition=" << repetition
+              << " producers=" << FLAGS_producer_threads << " operations=" << operations
+              << " elapsed_seconds=" << elapsed_seconds
               << " operations_per_sec=" << static_cast<double>(operations) / elapsed_seconds
               << " avg_us=" << latency.average_us << " p50_us=" << latency.p50_us
               << " p95_us=" << latency.p95_us << " p99_us=" << latency.p99_us
@@ -867,10 +917,12 @@ void print_index_result(std::string_view variant, size_t operations, double elap
 /// @param variant Stable output label.
 /// @param key_count Number of distinct cache hashes used by the workload.
 /// @param populate Whether every lookup should hit a pre-populated entry.
-Status run_index_case(std::string variant, size_t key_count, bool populate) {
+/// @param repetition One-based repetition index included in output.
+Status run_index_case(std::string variant, size_t key_count, bool populate, size_t repetition) {
     const size_t producer_count = static_cast<size_t>(FLAGS_producer_threads);
     const size_t operations_per_thread = static_cast<size_t>(FLAGS_index_operations_per_thread);
-    InflightWriteBufferIndex index(64, FLAGS_cache_path + "_index_" + variant);
+    InflightWriteBufferIndex index(
+            64, FLAGS_cache_path + "_index_" + variant + "_" + std::to_string(repetition));
     std::vector<UInt128Wrapper> hashes;
     hashes.reserve(key_count);
     const uint64_t epoch = 1;
@@ -918,7 +970,7 @@ Status run_index_case(std::string variant, size_t key_count, bool populate) {
     }
 
     const size_t total_operations = producer_count * operations_per_thread;
-    print_index_result(variant, total_operations,
+    print_index_result(variant, repetition, total_operations,
                        std::chrono::duration<double>(end - start).count(),
                        summarize_latencies(latencies));
     return Status::OK();
@@ -939,26 +991,33 @@ Status run_benchmarks(const std::vector<std::string>& modes,
     RETURN_IF_ERROR(environment.initialize(cache_capacity));
     std::cout << "CONFIG cache_path=" << FLAGS_cache_path << " cache_capacity=" << cache_capacity
               << " block_size=" << FLAGS_block_size << " request_size=" << FLAGS_request_size
-              << " producer_threads=" << FLAGS_producer_threads << '\n';
-
-    if (mode_enabled(modes, "reader")) {
-        RETURN_IF_ERROR(run_reader_case(&environment, CacheWriteMode::SYNC_WRITE, "sync_write"));
-        RETURN_IF_ERROR(run_reader_case(&environment, CacheWriteMode::ASYNC_WRITE, "async_write"));
-    }
-    if (mode_enabled(modes, "service")) {
-        for (size_t workers : worker_counts) {
+              << " producer_threads=" << FLAGS_producer_threads
+              << " repetitions=" << FLAGS_repetitions << '\n';
+
+    for (size_t repetition = 1; repetition <= FLAGS_repetitions; ++repetition) {
+        if (mode_enabled(modes, "reader")) {
+            RETURN_IF_ERROR(run_reader_case(&environment, CacheWriteMode::SYNC_WRITE, "sync_write",
+                                            repetition));
+            RETURN_IF_ERROR(run_reader_case(&environment, CacheWriteMode::ASYNC_WRITE,
+                                            "async_write", repetition));
+        }
+        if (mode_enabled(modes, "service")) {
+            for (size_t workers : worker_counts) {
+                RETURN_IF_ERROR(run_service_case(
+                        &environment, "drain_workers_" + std::to_string(workers), workers,
+                        static_cast<size_t>(FLAGS_service_operations + workers), repetition));
+            }
             RETURN_IF_ERROR(run_service_case(
-                    &environment, "drain_workers_" + std::to_string(workers), workers,
-                    static_cast<size_t>(FLAGS_service_operations + workers)));
+                    &environment, "backpressure", worker_counts.back(),
+                    std::min<size_t>(FLAGS_backpressure_pending_tasks, FLAGS_service_operations),
+                    repetition));
+        }
+        if (mode_enabled(modes, "index")) {
+            RETURN_IF_ERROR(
+                    run_index_case("sharded_miss", FLAGS_index_key_count, false, repetition));
+            RETURN_IF_ERROR(run_index_case("sharded_hit", FLAGS_index_key_count, true, repetition));
+            RETURN_IF_ERROR(run_index_case("hot_key_hit", 1, true, repetition));
         }
-        RETURN_IF_ERROR(run_service_case(
-                &environment, "backpressure", worker_counts.back(),
-                std::min<size_t>(FLAGS_backpressure_pending_tasks, FLAGS_service_operations)));
-    }
-    if (mode_enabled(modes, "index")) {
-        RETURN_IF_ERROR(run_index_case("sharded_miss", FLAGS_index_key_count, false));
-        RETURN_IF_ERROR(run_index_case("sharded_hit", FLAGS_index_key_count, true));
-        RETURN_IF_ERROR(run_index_case("hot_key_hit", 1, true));
     }
     return Status::OK();
 }
@@ -969,6 +1028,8 @@ Status run_benchmarks(const std::vector<std::string>& modes,
 int main(int argc, char** argv) {
     google::ParseCommandLineFlags(&argc, &argv, true);
     FLAGS_logtostderr = true;
+    // Keep RESULT lines parseable when the documented runner merges stdout and stderr.
+    FLAGS_minloglevel = google::GLOG_ERROR;
     google::InitGoogleLogging(argv[0]);
 
     // Doris configuration defaults contain ${DORIS_HOME}; a standalone benchmark has no launcher