[fix](be) Stop two SNII unit tests from trampling shared state (#66961)

### What problem does this PR solve?

Issue Number: N/A

Related PR: N/A

Problem Summary:

Two cases under `be/test/storage/index/snii/` reach outside their own
test and
break unrelated cases. Both are test-only; no production code changes.

**1. A fixed `/tmp` scratch path (`metered_file_reader_test.cpp`)**

`MakeRampFile()` built its 256-byte ramp file at the constant path
`/tmp/snii_metered_ramp.bin`. On a shared build machine that file
belongs to
whoever ran the suite first; every later user gets `EACCES` from
`::open`, and
all seven `SniiMeteredFileReader` cases fail at the first assertion with
no hint
that a foreign file is the cause:

```
metered_file_reader_test.cpp:42: Failure
Value of: w.open(path).ok()  Actual: false  Expected: true
```

Every other scratch path in this subtree already namespaces itself with
the pid
and a counter, and the nearest neighbour in the same directory
(`file_reader_caller_buffer_test.cpp`) also `std::remove()`s the file
afterwards.
This PR adopts both halves of that convention: the ramp file becomes a
`RampFile`
RAII object whose name carries the pid and a counter and whose
destructor removes
it. Removing matters as much as naming — a leftover file is what causes
the
collision in the first place, and pids get recycled.

**2. A permanently lowered `RLIMIT_NOFILE`
(`spimi_term_buffer_test.cpp`)**

`SniiSpimiTermBuffer.SpillOpenIoFailureLatched` drives a spill's
`::open` to
`EMFILE` by lowering `RLIMIT_NOFILE` to a hardcoded 64 and then opening
`/dev/null` until the table is full. Two things go wrong once the binary
already
holds more than 64 descriptors:

1. The cap assumes the process holds fewer than 64 descriptors, so the
hog loop
can still take one. Sampling `/proc/<pid>/fd` during a run shows the
SNII cases
alone keep the process at **14** descriptors, but adding the CLucene
inverted
index cases to the same binary takes it to **273** — they hold hundreds
of
index files open at once. Above the cap the very first `::open` already
fails,
`hogs` stays empty, and the case stops at `ASSERT_FALSE(hogs.empty())`.
What
   decides this is *which* cases share the binary, not how many.
2. That assertion is fatal, so it returns from the test body — and the
code that
restores `RLIMIT_NOFILE` sits after it and never runs. The whole process
stays
   capped at 64 descriptors for every later case.

The result is not a descriptor leak but a cascade: the next four
`SniiSpimiTermBuffer` cases fail instantly because they cannot open a
spill file,
and the first death test after them aborts the entire run when gtest's
`pipe()`
hits `EMFILE`:

```
CHECK failed: gtest-death-test.cc, line 1400: pipe(pipe_fd) != -1
Aborted
```

The fix holds the tight limit and the hog descriptors in an inner scope
whose
`Defer` restores `RLIMIT_NOFILE` and closes the descriptors on every
exit path,
fatal assertion included; and derives the cap from
`::dup(STDIN_FILENO)`, the
lowest free descriptor, instead of a fixed 64, so the case no longer
depends on
which tests ran before it. The case still drives the spill open to
`EMFILE` and
still asserts the error is latched.

**Before / after**, with

`--filter='*Snii*:*snii*:*SNII*:CollectionStatisticsTest.*:*InvertedIndex*:*inverted_index*'`:

| | before | after |
|---|---|---|
| `SniiMeteredFileReader` | 7 failed | pass |
| `SniiSpimiTermBuffer*` | 5 failed | pass |
| death test | aborted the run | pass |
| total | run truncated by the abort | 1693 passed, 0 failed |

be/test/storage/index/snii/io/metered_file_reader_test.cpp

@@ -18,8 +18,10 @@
 #include "storage/index/snii/io/metered_file_reader.h"
 
 #include <gtest/gtest.h>
+#include <unistd.h>
 
 #include <cstdint>
+#include <cstdio>
 #include <string>
 #include <vector>
 
@@ -35,27 +37,41 @@ using doris::snii::io::Range;
 
 namespace {
 
-// Writes 256 bytes (byte[i] = i) to a temp file and returns its path.
-std::string MakeRampFile() {
-    const std::string path = "/tmp/snii_metered_ramp.bin";
-    LocalFileWriter w;
-    EXPECT_TRUE(w.open(path).ok());
-    std::vector<uint8_t> data(256);
-    for (int i = 0; i < 256; ++i) {
-        data[i] = static_cast<uint8_t>(i);
+// Owns one scratch file holding 256 bytes (byte[i] = i) and removes it on scope
+// exit. The name carries the pid and a counter, like the other scratch paths in
+// this directory: a fixed /tmp name left behind by another user on a shared
+// machine makes ::open fail with EACCES and takes every test here down with it.
+class RampFile {
+public:
+    RampFile() {
+        static int counter = 0;
+        path_ = "/tmp/snii_metered_ramp_" + std::to_string(::getpid()) + "_" +
+                std::to_string(counter++) + ".bin";
+        LocalFileWriter w;
+        EXPECT_TRUE(w.open(path_).ok());
+        std::vector<uint8_t> data(256);
+        for (int i = 0; i < 256; ++i) {
+            data[i] = static_cast<uint8_t>(i);
+        }
+        EXPECT_TRUE(w.append(Slice(data)).ok());
+        EXPECT_TRUE(w.finalize().ok());
     }
-    EXPECT_TRUE(w.append(Slice(data)).ok());
-    EXPECT_TRUE(w.finalize().ok());
-    return path;
-}
+    ~RampFile() { std::remove(path_.c_str()); }
+
+    const std::string& path() const { return path_; }
+
+private:
+    std::string path_;
+};
 
 } // namespace
 
 // Single reads: first read to a block is a cache miss (1 round, 1 GET, 1 block of
 // remote bytes); a second read to the same 16-byte block is a hit (no new round).
 TEST(SniiMeteredFileReader, SingleReadCacheAccounting) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, /*block_size=*/16);
 
     std::vector<uint8_t> out;
@@ -84,8 +100,9 @@ TEST(SniiMeteredFileReader, SingleReadCacheAccounting) {
 
 // A read spanning 3 contiguous blocks is one round and one coalesced GET.
 TEST(SniiMeteredFileReader, SpanMultipleBlocksCoalesced) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, 16);
 
     std::vector<uint8_t> out;
@@ -98,8 +115,9 @@ TEST(SniiMeteredFileReader, SpanMultipleBlocksCoalesced) {
 
 // A batch of reads to non-adjacent blocks: one serial round, one GET per run.
 TEST(SniiMeteredFileReader, BatchNonAdjacent) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, 16);
 
     std::vector<Range> ranges = {{.offset = 0, .len = 4},
@@ -117,8 +135,9 @@ TEST(SniiMeteredFileReader, BatchNonAdjacent) {
 
 // A batch of reads to adjacent blocks coalesces into a single GET.
 TEST(SniiMeteredFileReader, BatchAdjacentCoalesced) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, 16);
 
     std::vector<Range> ranges = {{.offset = 0, .len = 4},
@@ -134,8 +153,9 @@ TEST(SniiMeteredFileReader, BatchAdjacentCoalesced) {
 
 // reset_metrics clears both counters and the resident cache (cold query).
 TEST(SniiMeteredFileReader, ResetClearsCacheAndCounters) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, 16);
 
     std::vector<uint8_t> out;
@@ -150,8 +170,9 @@ TEST(SniiMeteredFileReader, ResetClearsCacheAndCounters) {
 }
 
 TEST(SniiMeteredFileReader, InvalidRangeDoesNotPolluteMetrics) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, 16);
 
     std::vector<uint8_t> out;
@@ -165,8 +186,9 @@ TEST(SniiMeteredFileReader, InvalidRangeDoesNotPolluteMetrics) {
 }
 
 TEST(SniiMeteredFileReader, InvalidBatchRangeDoesNotPolluteMetrics) {
+    const RampFile ramp;
     LocalFileReader inner;
-    ASSERT_TRUE(inner.open(MakeRampFile()).ok());
+    ASSERT_TRUE(inner.open(ramp.path()).ok());
     MeteredFileReader m(&inner, 16);
 
     std::vector<std::vector<uint8_t>> outs;

be/test/storage/index/snii/writer/spimi_term_buffer_test.cpp

@@ -31,6 +31,7 @@
 
 #include "common/status.h"
 #include "storage/index/snii/writer/term_posting_test_utils.h"
+#include "util/defer_op.h"
 
 using doris::snii::writer::MemoryReporter;
 using doris::snii::writer::SpimiTermBuffer;
@@ -514,34 +515,46 @@ TEST(SniiSpimiTermBuffer, SpillOpenIoFailureLatched) {
     // Tiny threshold so the very first token triggers a spill_to_run().
     SpimiTermBuffer buf(/*has_positions=*/false, /*spill_threshold_bytes=*/1);
 
-    // Cap the soft limit low so we can exhaust the fd table cheaply, then hold it.
     struct rlimit saved {};
     ASSERT_EQ(getrlimit(RLIMIT_NOFILE, &saved), 0);
-    struct rlimit tight = saved;
-    tight.rlim_cur = 64; // small, but >= the few gtest/std fds already open
-    tight.rlim_cur = std::min(tight.rlim_cur, saved.rlim_max);
-    ASSERT_EQ(setrlimit(RLIMIT_NOFILE, &tight), 0);
+    {
+        std::vector<int> hogs;
+        // Restore on EVERY exit path, including a fatal assertion below. Leaving the
+        // process at the tight limit would break every later test in the binary, and
+        // gtest's death tests would abort the whole run when their pipe() hits EMFILE.
+        doris::Defer restore([&] {
+            for (int fd : hogs) {
+                ::close(fd);
+            }
+            EXPECT_EQ(setrlimit(RLIMIT_NOFILE, &saved), 0);
+        });
 
-    // Open /dev/null until the table is full: every free fd below the limit is now
-    // taken, so the next ::open (the spill's) cannot get one -> EMFILE.
-    std::vector<int> hogs;
-    for (;;) {
-        int fd = ::open("/dev/null", O_RDONLY);
-        if (fd < 0) {
-            break; // table exhausted
+        // Cap just above the descriptors this process already holds instead of a fixed
+        // number: how many are open depends on which tests ran first, and a cap below
+        // that leaves no free descriptor for the hog loop to take. dup() returns the
+        // lowest free descriptor, so every number under the cap is still available.
+        const int lowest_free = ::dup(STDIN_FILENO);
+        ASSERT_GE(lowest_free, 0);
+        ASSERT_EQ(::close(lowest_free), 0);
+        struct rlimit tight = saved;
+        tight.rlim_cur = std::min<rlim_t>(static_cast<rlim_t>(lowest_free) + 8, saved.rlim_max);
+        ASSERT_EQ(setrlimit(RLIMIT_NOFILE, &tight), 0);
+
+        // Open /dev/null until the table is full: every free fd below the limit is now
+        // taken, so the next ::open (the spill's) cannot get one -> EMFILE.
+        for (;;) {
+            int fd = ::open("/dev/null", O_RDONLY);
+            if (fd < 0) {
+                break; // table exhausted
+            }
+            hogs.push_back(fd);
         }
-        hogs.push_back(fd);
+        ASSERT_FALSE(hogs.empty());
+
+        buf.add_token("z", 0, 0); // triggers a spill whose RunWriter::open must fail
     }
-    ASSERT_FALSE(hogs.empty());
-
-    buf.add_token("z", 0, 0); // triggers a spill whose RunWriter::open must fail
-
-    // Release the hog fds and restore the limit before asserting (so gtest I/O works).
-    for (int fd : hogs) {
-        ::close(fd);
-    }
-    ASSERT_EQ(setrlimit(RLIMIT_NOFILE, &saved), 0);
-
+    // The hog fds are released and the limit restored here, before asserting, so gtest
+    // I/O works.
     EXPECT_FALSE(buf.status().ok()) << "spill open() failure must latch an error";
 }