[Data] Always print traceback for internal exceptions (#46647)

If Ray Data raises an internal error (not from the user-provided UDF),
then the traceback is mostly hidden.
```
2024-07-15 20:45:45,956 ERROR exceptions.py:73 -- Exception occurred in Ray Data or Ray Core internal code. If you continue to see this error, please open an issue on the Ray project GitHub page with the full stack trace below: https://github.com/ray-project/ray/issues/new/choose
ray.data.exceptions.SystemException

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/balaji/Documents/GitHub/ray/python/ray/data/tests/1.py", line 3, in <module>
    ray.data.range(1).materialize()
  File "/Users/balaji/Documents/GitHub/ray/python/ray/data/dataset.py", line 4599, in materialize
    copy._plan.execute()
  File "/Users/balaji/Documents/GitHub/ray/python/ray/data/exceptions.py", line 90, in handle_trace
    raise e.with_traceback(None) from SystemException()
RuntimeError: Mock internal error
```

This output isn't helpful for debugging, so this PR changes the behavior
so that the full traceback is printed for all internal errors.

Signed-off-by: Balaji Veeramani <[email protected]>

python/ray/data/exceptions.py

@@ -77,8 +77,11 @@ def handle_trace(*args, **kwargs):
                     "https://github.com/ray-project/ray/issues/new/choose"
                 )
 
+            should_hide_traceback = is_user_code_exception and not log_to_stdout
             logger.exception(
-                "Full stack trace:", exc_info=True, extra={"hide": not log_to_stdout}
+                "Full stack trace:",
+                exc_info=True,
+                extra={"hide": should_hide_traceback},
             )
             if is_user_code_exception:
                 raise e.with_traceback(None)

python/ray/data/tests/test_exceptions.py

@@ -9,7 +9,17 @@
 from ray.tests.conftest import *  # noqa
 
 
-def test_user_exception(caplog, propagate_logs, ray_start_regular_shared):
+@pytest.mark.parametrize("log_internal_stack_trace_to_stdout", [True, False])
+def test_user_exception(
+    log_internal_stack_trace_to_stdout,
+    caplog,
+    propagate_logs,
+    restore_data_context,
+    ray_start_regular_shared,
+):
+    ctx = ray.data.DataContext.get_current()
+    ctx.log_internal_stack_trace_to_stdout = log_internal_stack_trace_to_stdout
+
     def f(row):
         1 / 0
         return row
@@ -21,16 +31,17 @@ def f(row):
     assert issubclass(exc_info.type, UserCodeException)
     assert ZeroDivisionError.__name__ in str(exc_info.value)
 
-    assert any(
-        record.levelno == logging.ERROR
-        and "Exception occurred in user code" in record.message
-        for record in caplog.records
-    ), caplog.records
+    if not log_internal_stack_trace_to_stdout:
+        assert any(
+            record.levelno == logging.ERROR
+            and "Exception occurred in user code" in record.message
+            for record in caplog.records
+        ), caplog.records
 
     assert any(
         record.levelno == logging.ERROR
         and "Full stack trace:" in record.message
-        and record.hide
+        and getattr(record, "hide", False) == (not log_internal_stack_trace_to_stdout)
         for record in caplog.records
     ), caplog.records
 
@@ -58,7 +69,7 @@ class FakeException(Exception):
     assert any(
         record.levelno == logging.ERROR
         and "Full stack trace:" in record.message
-        and record.hide
+        and not getattr(record, "hide", False)
         for record in caplog.records
     ), caplog.records
 
@@ -80,7 +91,9 @@ def f(row):
     assert ZeroDivisionError.__name__ in str(exc_info.value)
 
     assert any(
-        record.levelno == logging.ERROR and "Full stack trace:" in record.message
+        record.levelno == logging.ERROR
+        and "Full stack trace:" in record.message
+        and not getattr(record, "hide", False)
         for record in caplog.records
     ), caplog.records