chore: replace blanket S301 allow with per-line noqa at pickle.loads

The pickle-Expr work added `S301` (suspicious-pickle-usage) to the
`per-file-ignores` for both `python/tests/*` and `examples/*`. That
silenced ruff's pickle-usage warning everywhere in those trees, not
just at the call sites that genuinely need it — a future test that
adds an unrelated `pickle.loads` on untrusted bytes would pass review
without ever pinging the linter.

Drop the blanket allows and tag each intentional `pickle.loads` call
with a per-line `# noqa: S301` instead (12 sites total: 9 in
`test_pickle_expr.py`, 2 in `_pickle_multiprocessing_helpers.py`,
1 in the FFI example test). New `pickle.loads` calls now have to
justify themselves with their own noqa, keeping S301 a real signal.

Production code (`Expr.__reduce__`) only calls `pickle.dumps`, which
S301 does not cover, so no noqa is needed outside tests/examples.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: timsaucer

examples/datafusion-ffi-example/python/tests/_test_pickle_strict_ffi.py

@@ -98,7 +98,7 @@ def test_strict_ffi_udf_pickle_roundtrip_via_thread_locals():
 
     set_worker_ctx(receiver)
     try:
-        restored = pickle.loads(blob)
+        restored = pickle.loads(blob)  # noqa: S301
     finally:
         clear_worker_ctx()
 

pyproject.toml

@@ -127,7 +127,6 @@ extend-allowed-calls = ["datafusion.lit", "lit"]
   "PT011",
   "RUF015",
   "S101",
-  "S301",
   "S608",
   "SLF",
 ]
@@ -141,7 +140,6 @@ extend-allowed-calls = ["datafusion.lit", "lit"]
   "PLR2004",
   "RUF015",
   "S101",
-  "S301",
   "T201",
   "W505",
 ]

python/tests/_pickle_multiprocessing_helpers.py

@@ -69,7 +69,7 @@ def unpickle_and_describe(blob: bytes) -> str:
     """Unpickle a proto-bytes blob and return its canonical name."""
     import pickle
 
-    expr = pickle.loads(blob)
+    expr = pickle.loads(blob)  # noqa: S301
     return expr.canonical_name()
 
 
@@ -82,7 +82,7 @@ def unpickle_and_evaluate(blob: bytes, batch: list[int]) -> list[int]:
     """
     import pickle
 
-    expr = pickle.loads(blob)
+    expr = pickle.loads(blob)  # noqa: S301
     ctx = SessionContext()
     df = ctx.from_pydict({"a": batch})
     out = df.with_column("result", expr).select("result")

python/tests/test_pickle_expr.py

@@ -67,7 +67,7 @@ class TestProtoRoundTrip:
     def test_builtin_round_trip(self):
         e = col("a") + lit(1)
         blob = pickle.dumps(e)
-        decoded = pickle.loads(blob)
+        decoded = pickle.loads(blob)  # noqa: S301
         assert decoded.canonical_name() == e.canonical_name()
 
     def test_to_bytes_from_bytes(self):
@@ -107,14 +107,14 @@ def test_udf_decodes_into_fresh_ctx(self):
     def test_udf_decodes_via_pickle_with_no_worker_ctx(self):
         e = _double_udf()(col("a"))
         blob = pickle.dumps(e)
-        decoded = pickle.loads(blob)
+        decoded = pickle.loads(blob)  # noqa: S301
         assert "double" in decoded.canonical_name()
 
     def test_udf_decodes_via_pickle_with_worker_ctx(self):
         set_worker_ctx(SessionContext())
         e = _double_udf()(col("a"))
         blob = pickle.dumps(e)
-        decoded = pickle.loads(blob)
+        decoded = pickle.loads(blob)  # noqa: S301
         assert "double" in decoded.canonical_name()
 
     def test_closure_capturing_udf_names_match(self):
@@ -132,7 +132,7 @@ def fn(arr):
         )
         e = u(col("a"))
         blob = pickle.dumps(e)
-        decoded = pickle.loads(blob)
+        decoded = pickle.loads(blob)  # noqa: S301
         # Round-trip names match; functional verification of captured state
         # happens in test_pickle_multiprocessing via an actual UDF call.
         assert decoded.canonical_name() == e.canonical_name()
@@ -189,15 +189,15 @@ def test_agg_udf_decodes_via_pickle_with_no_worker_ctx(self):
         u = self._build_aggregate_udf()
         e = u(col("a"))
         blob = pickle.dumps(e)
-        decoded = pickle.loads(blob)
+        decoded = pickle.loads(blob)  # noqa: S301
         assert "count_all" in decoded.canonical_name()
 
     def test_agg_udf_evaluates_after_roundtrip(self):
         """End-to-end: the decoded aggregate UDF runs and merges across
         partitions, exercising the round-tripped state-field schema."""
         u = self._build_aggregate_udf()
         e = u(col("a"))
-        decoded = pickle.loads(pickle.dumps(e))
+        decoded = pickle.loads(pickle.dumps(e))  # noqa: S301
 
         ctx = SessionContext()
         df = ctx.from_pydict({"a": [1, 2, 3, 4, 5]})
@@ -242,7 +242,7 @@ def test_window_udf_decodes_via_pickle_with_no_worker_ctx(self):
         u = self._build_window_udf()
         e = u(col("a"))
         blob = pickle.dumps(e)
-        decoded = pickle.loads(blob)
+        decoded = pickle.loads(blob)  # noqa: S301
         assert "count_up" in decoded.canonical_name()
 
 
@@ -344,7 +344,7 @@ def test_sender_ctx_strict_roundtrip_via_pickle(self):
         worker.register_udf(u)
         set_worker_ctx(worker)
         try:
-            decoded = pickle.loads(blob)
+            decoded = pickle.loads(blob)  # noqa: S301
         finally:
             clear_worker_ctx()
 
@@ -369,7 +369,7 @@ def test_sender_ctx_strict_pickle_accepted_by_inline_worker_with_registry(self):
         worker.register_udf(u)
         set_worker_ctx(worker)
         try:
-            decoded = pickle.loads(blob)
+            decoded = pickle.loads(blob)  # noqa: S301
         finally:
             clear_worker_ctx()