Merge pull request ClickHouse#91895 from ClickHouse/backport/25.8/91750

Backport ClickHouse#91750 to 25.8: Fix `too many open files` errors in integration tests

Author: clickhouse-gh[bot]

tests/integration/helpers/client.py

@@ -181,9 +181,9 @@ def __init__(
         self, command, stdin=None, timeout=None, ignore_error=False, parse=False, stdout_file_path=None, stderr_file_path=None, env = {}
     ):
         # Write data to tmp file to avoid PIPEs and execution blocking
-        stdin_file = tempfile.TemporaryFile(mode="w+")
-        stdin_file.write(stdin)
-        stdin_file.seek(0)
+        self.stdin_file = tempfile.TemporaryFile(mode="w+")
+        self.stdin_file.write(stdin)
+        self.stdin_file.seek(0)
         self.stdout_file = tempfile.TemporaryFile() if stdout_file_path is None else stdout_file_path
         self.stderr_file = tempfile.TemporaryFile() if stderr_file_path is None else stderr_file_path
         self.ignore_error = ignore_error
@@ -196,7 +196,7 @@ def __init__(
         env["TSAN_OPTIONS"] = "use_sigaltstack=0 verbosity=0"
         self.process = sp.Popen(
             command,
-            stdin=stdin_file,
+            stdin=self.stdin_file,
             stdout=self.stdout_file,
             stderr=self.stderr_file,
             env=env,
@@ -225,13 +225,24 @@ def remove_trash_from_stderr(self, stderr):
         ]
         return "\n".join(lines)
 
-    def get_answer(self):
-        self.process.wait(timeout=DEFAULT_QUERY_TIMEOUT)
-        self.stdout_file.seek(0)
-        self.stderr_file.seek(0)
+    def wait_and_read_output(self):
+        try:
+            self.process.wait(timeout=DEFAULT_QUERY_TIMEOUT)
+            self.stdout_file.seek(0)
+            self.stderr_file.seek(0)
+
+            stdout = self.stdout_file.read().decode("utf-8", errors="replace")
+            stderr = self.stderr_file.read().decode("utf-8", errors="replace")
 
-        stdout = self.stdout_file.read().decode("utf-8", errors="replace")
-        stderr = self.stderr_file.read().decode("utf-8", errors="replace")
+            return stdout, stderr
+
+        finally:
+            self.stdin_file.close()
+            self.stdout_file.close()
+            self.stderr_file.close()
+
+    def get_answer(self):
+        stdout, stderr = self.wait_and_read_output()
 
         if (
             self.timer is not None
@@ -264,12 +275,7 @@ def get_answer(self):
         return stdout
 
     def get_error(self):
-        self.process.wait(timeout=DEFAULT_QUERY_TIMEOUT)
-        self.stdout_file.seek(0)
-        self.stderr_file.seek(0)
-
-        stdout = self.stdout_file.read().decode("utf-8", errors="replace")
-        stderr = self.stderr_file.read().decode("utf-8", errors="replace")
+        stdout, stderr = self.wait_and_read_output()
 
         if (
             self.timer is not None
@@ -288,12 +294,7 @@ def get_error(self):
         return stderr
 
     def get_answer_and_error(self):
-        self.process.wait(timeout=DEFAULT_QUERY_TIMEOUT)
-        self.stdout_file.seek(0)
-        self.stderr_file.seek(0)
-
-        stdout = self.stdout_file.read().decode("utf-8", errors="replace")
-        stderr = self.stderr_file.read().decode("utf-8", errors="replace")
+        stdout, stderr = self.wait_and_read_output()
 
         if (
             self.timer is not None