Refactor run query output handling (#1373)

Co-authored-by: Thomas Kunwar <[email protected]>

Author: dreadatour

src/datachain/catalog/catalog.py

@@ -133,19 +133,26 @@ def shutdown_process(
             return proc.wait()
 
 
-def _process_stream(stream: "IO[bytes]", callback: Callable[[str], None]) -> None:
+def process_output(stream: IO[bytes], callback: Callable[[str], None]) -> None:
     buffer = b""
-    while byt := stream.read(1):  # Read one byte at a time
-        buffer += byt
 
-        if byt in (b"\n", b"\r"):  # Check for newline or carriage return
-            line = buffer.decode("utf-8")
-            callback(line)
-            buffer = b""  # Clear buffer for next line
+    try:
+        while byt := stream.read(1):  # Read one byte at a time
+            buffer += byt
 
-    if buffer:  # Handle any remaining data in the buffer
-        line = buffer.decode("utf-8")
-        callback(line)
+            if byt in (b"\n", b"\r"):  # Check for newline or carriage return
+                line = buffer.decode("utf-8", errors="replace")
+                callback(line)
+                buffer = b""  # Clear buffer for the next line
+
+        if buffer:  # Handle any remaining data in the buffer
+            line = buffer.decode("utf-8", errors="replace")
+            callback(line)
+    finally:
+        try:
+            stream.close()  # Ensure output is closed
+        except Exception:  # noqa: BLE001, S110
+            pass
 
 
 class DatasetRowsFetcher(NodesThreadPool):
@@ -1747,13 +1754,13 @@ def clone(
             recursive=recursive,
         )
 
+    @staticmethod
     def query(
-        self,
         query_script: str,
         env: Mapping[str, str] | None = None,
         python_executable: str = sys.executable,
-        capture_output: bool = False,
-        output_hook: Callable[[str], None] = noop,
+        stdout_callback: Callable[[str], None] | None = None,
+        stderr_callback: Callable[[str], None] | None = None,
         params: dict[str, str] | None = None,
         job_id: str | None = None,
         reset: bool = False,
@@ -1773,13 +1780,18 @@ def query(
             },
         )
         popen_kwargs: dict[str, Any] = {}
-        if capture_output:
-            popen_kwargs = {"stdout": subprocess.PIPE, "stderr": subprocess.STDOUT}
+
+        if stdout_callback is not None:
+            popen_kwargs = {"stdout": subprocess.PIPE}
+        if stderr_callback is not None:
+            popen_kwargs["stderr"] = subprocess.PIPE
 
         def raise_termination_signal(sig: int, _: Any) -> NoReturn:
             raise TerminationSignal(sig)
 
-        thread: Thread | None = None
+        stdout_thread: Thread | None = None
+        stderr_thread: Thread | None = None
+
         with subprocess.Popen(cmd, env=env, **popen_kwargs) as proc:  # noqa: S603
             logger.info("Starting process %s", proc.pid)
 
@@ -1793,10 +1805,20 @@ def raise_termination_signal(sig: int, _: Any) -> NoReturn:
             orig_sigterm_handler = signal.getsignal(signal.SIGTERM)
             signal.signal(signal.SIGTERM, raise_termination_signal)
             try:
-                if capture_output:
-                    args = (proc.stdout, output_hook)
-                    thread = Thread(target=_process_stream, args=args, daemon=True)
-                    thread.start()
+                if stdout_callback is not None:
+                    stdout_thread = Thread(
+                        target=process_output,
+                        args=(proc.stdout, stdout_callback),
+                        daemon=True,
+                    )
+                    stdout_thread.start()
+                if stderr_callback is not None:
+                    stderr_thread = Thread(
+                        target=process_output,
+                        args=(proc.stderr, stderr_callback),
+                        daemon=True,
+                    )
+                    stderr_thread.start()
 
                 proc.wait()
             except TerminationSignal as exc:
@@ -1814,8 +1836,22 @@ def raise_termination_signal(sig: int, _: Any) -> NoReturn:
             finally:
                 signal.signal(signal.SIGTERM, orig_sigterm_handler)
                 signal.signal(signal.SIGINT, orig_sigint_handler)
-                if thread:
-                    thread.join()  # wait for the reader thread
+                # wait for the reader thread
+                thread_join_timeout_seconds = 30
+                if stdout_thread is not None:
+                    stdout_thread.join(timeout=thread_join_timeout_seconds)
+                    if stdout_thread.is_alive():
+                        logger.warning(
+                            "stdout thread is still alive after %s seconds",
+                            thread_join_timeout_seconds,
+                        )
+                if stderr_thread is not None:
+                    stderr_thread.join(timeout=thread_join_timeout_seconds)
+                    if stderr_thread.is_alive():
+                        logger.warning(
+                            "stderr thread is still alive after %s seconds",
+                            thread_join_timeout_seconds,
+                        )
 
         logger.info("Process %s exited with return code %s", proc.pid, proc.returncode)
         if proc.returncode in (

tests/unit/test_query.py

@@ -53,12 +53,31 @@ def test_args(catalog, mock_popen, reset):
     mock_popen.assert_called_once_with(["mypython", "-c", "pass"], env=expected_env)
 
 
+def test_capture_stdout(catalog, mock_popen):
+    mock_popen.stdout = io.BytesIO(b"Hello, World!\rLorem Ipsum\nDolor Sit Amet\nconse")
+    stdout = []
+
+    catalog.query("pass", stdout_callback=stdout.append)
+    assert stdout == ["Hello, World!\r", "Lorem Ipsum\n", "Dolor Sit Amet\n", "conse"]
+
+
+def test_capture_stderr(catalog, mock_popen):
+    mock_popen.stderr = io.BytesIO(b"Hello, World!\rLorem Ipsum\nDolor Sit Amet\nconse")
+    stderr = []
+
+    catalog.query("pass", stderr_callback=stderr.append)
+    assert stderr == ["Hello, World!\r", "Lorem Ipsum\n", "Dolor Sit Amet\n", "conse"]
+
+
 def test_capture_output(catalog, mock_popen):
     mock_popen.stdout = io.BytesIO(b"Hello, World!\rLorem Ipsum\nDolor Sit Amet\nconse")
-    lines = []
+    mock_popen.stderr = io.BytesIO(b"foo\nbar")
+    stdout = []
+    stderr = []
 
-    catalog.query("pass", capture_output=True, output_hook=lines.append)
-    assert lines == ["Hello, World!\r", "Lorem Ipsum\n", "Dolor Sit Amet\n", "conse"]
+    catalog.query("pass", stdout_callback=stdout.append, stderr_callback=stderr.append)
+    assert stdout == ["Hello, World!\r", "Lorem Ipsum\n", "Dolor Sit Amet\n", "conse"]
+    assert stderr == ["foo\n", "bar"]
 
 
 def test_canceled_by_user(catalog, mock_popen):