[SPARK-54323][PYTHON] Change the way to access logs to TVF instead of system view

### What changes were proposed in this pull request?

Changes the way to access logs to TVF instead of system view.

```sql
SELECT * FROM python_worker_logs()
```

```py
spark.tvf.python_worker_logs()
```

Also blocks the TVF when the python worker logging is disabled.

```py
>>> spark.conf.get('spark.sql.pyspark.worker.logging.enabled')
'false'
>>> spark.tvf.python_worker_logs().show()
Traceback (most recent call last):
...
pyspark.errors.exceptions.captured.AnalysisException: [FEATURE_NOT_ENABLED] The feature Python Worker Logging is not enabled. Consider setting the config spark.sql.pyspark.worker.logging.enabled to true to enable this capability. SQLSTATE: 56038

>>> spark.conf.set('spark.sql.pyspark.worker.logging.enabled', True)
>>> spark.tvf.python_worker_logs().show()
+---+-----+---+-------+---------+------+
| ts|level|msg|context|exception|logger|
+---+-----+---+-------+---------+------+
+---+-----+---+-------+---------+------+
```

### Why are the changes needed?

There may be namespace conflicts with the other system tables/views, etc..

For example, the variables in SQL has the same namespace `system.session.varname`, which may potentially cause an issue.

### Does this PR introduce _any_ user-facing change?

Yes, the way to access python worker logs will be changed.

### How was this patch tested?

Modified the related tests.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #53021 from ueshin/issues/SPARK-54323/tvf.

Authored-by: Takuya Ueshin <ueshin@databricks.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: ueshin

project/MimaExcludes.scala

@@ -52,7 +52,10 @@ object MimaExcludes {
     ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.Dataset.repartitionById"),
 
     // [SPARK-54001][CONNECT] Replace block copying with ref-counting in ArtifactManager cloning
-    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.artifact.ArtifactManager.cachedBlockIdList")
+    ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.artifact.ArtifactManager.cachedBlockIdList"),
+
+    // [SPARK-54323][PYTHON] Change the way to access logs to TVF instead of system view
+    ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.TableValuedFunction.python_worker_logs")
   )
 
   // Default exclude rules

python/docs/source/user_guide/bugbusting.ipynb

@@ -945,7 +945,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "logs = spark.table(\"system.session.python_worker_logs\")"
+    "logs = spark.tvf.python_worker_logs()"
    ]
   },
   {
@@ -1021,7 +1021,7 @@
     "df.select(my_udf(\"text\")).show()\n",
     "\n",
     "# Query the logs\n",
-    "logs = spark.table(\"system.session.python_worker_logs\")\n",
+    "logs = spark.tvf.python_worker_logs()\n",
     "logs.select(\"level\", \"msg\", \"logger\", \"context\").show(truncate=False)"
    ]
   },
@@ -1076,7 +1076,7 @@
     "spark.conf.set(\"spark.sql.pyspark.worker.logging.enabled\", \"true\")\n",
     "spark.range(1).select(contextual_udf(lit(\"test\"))).show()\n",
     "\n",
-    "logs = spark.table(\"system.session.python_worker_logs\")\n",
+    "logs = spark.tvf.python_worker_logs()\n",
     "logs.filter(\"logger = 'contextual'\").select(\"msg\", \"context\").show(truncate=False)"
    ]
   },
@@ -1135,7 +1135,7 @@
     "spark.conf.set(\"spark.sql.pyspark.worker.logging.enabled\", \"true\")\n",
     "spark.createDataFrame([(0,), (5,)], [\"value\"]).select(failing_udf(\"value\")).show()\n",
     "\n",
-    "logs = spark.table(\"system.session.python_worker_logs\")\n",
+    "logs = spark.tvf.python_worker_logs()\n",
     "logs.filter(\"logger = 'error_handler'\").select(\"msg\", \"exception\").show(truncate=False)"
    ]
   },
@@ -1193,7 +1193,7 @@
     "df = spark.createDataFrame([(\"hello world\",)], [\"text\"])\n",
     "df.lateralJoin(WordSplitter(col(\"text\").outer())).show()\n",
     "\n",
-    "logs = spark.table(\"system.session.python_worker_logs\")\n",
+    "logs = spark.tvf.python_worker_logs()\n",
     "logs.filter(\"logger = 'udtf_logger'\").select(\"msg\", \"context\").show(truncate=False)"
    ]
   },
@@ -1231,7 +1231,7 @@
     }
    ],
    "source": [
-    "logs = spark.table(\"system.session.python_worker_logs\")\n",
+    "logs = spark.tvf.python_worker_logs()\n",
     "\n",
     "# Count logs by level\n",
     "logs.groupBy(\"level\").count().show()\n",

python/pyspark/sql/connect/tvf.py

@@ -109,6 +109,11 @@ def variant_explode_outer(self, input: "Column") -> "DataFrame":
 
     variant_explode_outer.__doc__ = PySparkTableValuedFunction.variant_explode_outer.__doc__
 
+    def python_worker_logs(self) -> "DataFrame":
+        return self._fn("python_worker_logs")
+
+    python_worker_logs.__doc__ = PySparkTableValuedFunction.python_worker_logs.__doc__
+
     def _fn(self, name: str, *args: "Column") -> "DataFrame":
         from pyspark.sql.connect.dataframe import DataFrame
         from pyspark.sql.connect.plan import UnresolvedTableValuedFunction

python/pyspark/sql/tests/arrow/test_arrow_cogrouped_map.py

@@ -396,20 +396,20 @@ def func_with_logging(left, right):
                 + [Row(id=2, v1=20, v2=200)],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
-
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"arrow cogrouped map: {dict(v1=v1, v2=v2)}",
-                    context={"func_name": func_with_logging.__name__},
-                    logger="test_arrow_cogrouped_map",
-                )
-                for v1, v2 in [([10, 30], [100, 300]), ([20], [200])]
-            ],
-        )
+            logs = self.spark.tvf.python_worker_logs()
+
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"arrow cogrouped map: {dict(v1=v1, v2=v2)}",
+                        context={"func_name": func_with_logging.__name__},
+                        logger="test_arrow_cogrouped_map",
+                    )
+                    for v1, v2 in [([10, 30], [100, 300]), ([20], [200])]
+                ],
+            )
 
 
 class CogroupedMapInArrowTests(CogroupedMapInArrowTestsMixin, ReusedSQLTestCase):

python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py

@@ -416,20 +416,20 @@ def func_with_logging(group):
                 df,
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
-
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"arrow grouped map: {dict(id=lst, value=[v*10 for v in lst])}",
-                    context={"func_name": func_with_logging.__name__},
-                    logger="test_arrow_grouped_map",
-                )
-                for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
-            ],
-        )
+            logs = self.spark.tvf.python_worker_logs()
+
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"arrow grouped map: {dict(id=lst, value=[v*10 for v in lst])}",
+                        context={"func_name": func_with_logging.__name__},
+                        logger="test_arrow_grouped_map",
+                    )
+                    for lst in [[0, 2, 4, 6, 8], [1, 3, 5, 7]]
+                ],
+            )
 
     @unittest.skipIf(is_remote_only(), "Requires JVM access")
     def test_apply_in_arrow_iter_with_logging(self):
@@ -456,20 +456,20 @@ def func_with_logging(group: Iterator[pa.RecordBatch]) -> Iterator[pa.RecordBatc
                 df,
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
+            logs = self.spark.tvf.python_worker_logs()
 
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"arrow grouped map: {dict(id=lst, value=[v*10 for v in lst])}",
-                    context={"func_name": func_with_logging.__name__},
-                    logger="test_arrow_grouped_map",
-                )
-                for lst in [[0, 2, 4], [6, 8], [1, 3, 5], [7]]
-            ],
-        )
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"arrow grouped map: {dict(id=lst, value=[v*10 for v in lst])}",
+                        context={"func_name": func_with_logging.__name__},
+                        logger="test_arrow_grouped_map",
+                    )
+                    for lst in [[0, 2, 4], [6, 8], [1, 3, 5], [7]]
+                ],
+            )
 
 
 class ApplyInArrowTests(ApplyInArrowTestsMixin, ReusedSQLTestCase):

python/pyspark/sql/tests/arrow/test_arrow_map.py

@@ -247,12 +247,12 @@ def func_with_logging(iterator):
                 [Row(id=i) for i in range(9)],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
+            logs = self.spark.tvf.python_worker_logs()
 
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            self._expected_logs_for_test_map_in_arrow_with_logging(func_with_logging.__name__),
-        )
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                self._expected_logs_for_test_map_in_arrow_with_logging(func_with_logging.__name__),
+            )
 
     def _expected_logs_for_test_map_in_arrow_with_logging(self, func_name):
         return [

python/pyspark/sql/tests/arrow/test_arrow_udf_grouped_agg.py

@@ -1044,20 +1044,20 @@ def my_grouped_agg_arrow_udf(x):
                 [Row(id=1, result=3.0), Row(id=2, result=18.0)],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
+            logs = self.spark.tvf.python_worker_logs()
 
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"grouped agg arrow udf: {n}",
-                    context={"func_name": my_grouped_agg_arrow_udf.__name__},
-                    logger="test_grouped_agg_arrow",
-                )
-                for n in [2, 3]
-            ],
-        )
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"grouped agg arrow udf: {n}",
+                        context={"func_name": my_grouped_agg_arrow_udf.__name__},
+                        logger="test_grouped_agg_arrow",
+                    )
+                    for n in [2, 3]
+                ],
+            )
 
 
 class GroupedAggArrowUDFTests(GroupedAggArrowUDFTestsMixin, ReusedSQLTestCase):

python/pyspark/sql/tests/arrow/test_arrow_udf_scalar.py

@@ -1201,20 +1201,20 @@ def my_scalar_arrow_udf(x):
                 [Row(result=f"scalar_arrow_{i}") for i in range(3)],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
+            logs = self.spark.tvf.python_worker_logs()
 
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"scalar arrow udf: {lst}",
-                    context={"func_name": my_scalar_arrow_udf.__name__},
-                    logger="test_scalar_arrow",
-                )
-                for lst in [[0], [1, 2]]
-            ],
-        )
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"scalar arrow udf: {lst}",
+                        context={"func_name": my_scalar_arrow_udf.__name__},
+                        logger="test_scalar_arrow",
+                    )
+                    for lst in [[0], [1, 2]]
+                ],
+            )
 
     @unittest.skipIf(is_remote_only(), "Requires JVM access")
     def test_scalar_iter_arrow_udf_with_logging(self):
@@ -1241,20 +1241,20 @@ def my_scalar_iter_arrow_udf(it):
                 [Row(result=f"scalar_iter_arrow_{i}") for i in range(9)],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
+            logs = self.spark.tvf.python_worker_logs()
 
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"scalar iter arrow udf: {lst}",
-                    context={"func_name": my_scalar_iter_arrow_udf.__name__},
-                    logger="test_scalar_iter_arrow",
-                )
-                for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
-            ],
-        )
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"scalar iter arrow udf: {lst}",
+                        context={"func_name": my_scalar_iter_arrow_udf.__name__},
+                        logger="test_scalar_iter_arrow",
+                    )
+                    for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
+                ],
+            )
 
 
 class ScalarArrowUDFTests(ScalarArrowUDFTestsMixin, ReusedSQLTestCase):

python/pyspark/sql/tests/arrow/test_arrow_udf_window.py

@@ -834,20 +834,20 @@ def my_window_arrow_udf(x):
                 ],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
-
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"window arrow udf: {lst}",
-                    context={"func_name": my_window_arrow_udf.__name__},
-                    logger="test_window_arrow",
-                )
-                for lst in [[1.0], [1.0, 2.0], [3.0], [3.0, 5.0], [3.0, 5.0, 10.0]]
-            ],
-        )
+            logs = self.spark.tvf.python_worker_logs()
+
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"window arrow udf: {lst}",
+                        context={"func_name": my_window_arrow_udf.__name__},
+                        logger="test_window_arrow",
+                    )
+                    for lst in [[1.0], [1.0, 2.0], [3.0], [3.0, 5.0], [3.0, 5.0, 10.0]]
+                ],
+            )
 
 
 class WindowArrowUDFTests(WindowArrowUDFTestsMixin, ReusedSQLTestCase):

python/pyspark/sql/tests/arrow/test_arrow_udtf.py

@@ -1721,20 +1721,20 @@ def eval(self, table_data: "pa.RecordBatch") -> Iterator["pa.Table"]:
                 [Row(id=i, doubled=i * 2) for i in range(9)],
             )
 
-        logs = self.spark.table("system.session.python_worker_logs")
+            logs = self.spark.tvf.python_worker_logs()
 
-        assertDataFrameEqual(
-            logs.select("level", "msg", "context", "logger"),
-            [
-                Row(
-                    level="WARNING",
-                    msg=f"arrow udtf: {dict(id=lst)}",
-                    context={"class_name": "TestArrowUDTFWithLogging", "func_name": "eval"},
-                    logger="test_arrow_udtf",
-                )
-                for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
-            ],
-        )
+            assertDataFrameEqual(
+                logs.select("level", "msg", "context", "logger"),
+                [
+                    Row(
+                        level="WARNING",
+                        msg=f"arrow udtf: {dict(id=lst)}",
+                        context={"class_name": "TestArrowUDTFWithLogging", "func_name": "eval"},
+                        logger="test_arrow_udtf",
+                    )
+                    for lst in [[0, 1, 2], [3], [4, 5, 6], [7, 8]]
+                ],
+            )
 
 
 class ArrowUDTFTests(ArrowUDTFTestsMixin, ReusedSQLTestCase):