[SPARK-59032][INFRA] Run behavioral e2e in a separate job so the parity gate gets a clean Connect server

.github/workflows/build_python_connect.yml

@@ -173,19 +173,10 @@ jobs:
             sleep 1
           done
 
-      - name: Rust behavioral e2e tests (against the live server)
-        # These assert results against a live server and otherwise self-gate on
-        # SPARK_REMOTE, so they no-op in the server-less rust.yml job. Run them here,
-        # where a Spark 4.2.0 Connect server is already up, so the drop-in behavior
-        # (functions/columns/dataframe/tvf/catalog/session/streaming/merge/ml/artifact)
-        # is actually verified in CI. Serial: they share the one server session.
-        env:
-          SPARK_REMOTE: sc://localhost:15002
-        run: |
-          cargo test -p apache-spark-connect \
-            --test e2e_functions --test e2e_columns --test e2e_dataframe \
-            --test e2e_coverage --test e2e_integration \
-            -- --test-threads=1
+      # NOTE (SPARK-59032): the behavioral e2e tests used to run here, sharing this
+      # server with the parity gate. They now run in their own job (connect-e2e,
+      # below) so the parity run starts against a server nothing else has touched.
+      # No test is skipped.
 
       - name: Parity - official connect suite through OUR Rust client
         id: parity
@@ -224,6 +215,107 @@ jobs:
             /tmp/connect-server.log
           if-no-files-found: warn
 
+  # ---------------------------------------------------------------------------
+  # Behavioral e2e tests (Rust drop-in against a live server), in their OWN job
+  # with a separate Connect server so they never share server state with the
+  # parity gate. The parity gate then starts against a server nothing else has
+  # touched, which avoids the SPARK-59032 arrow-UDF hang (a UDF that crashes a
+  # Python worker on the shared server wedged worker reuse for later tests).
+  # ---------------------------------------------------------------------------
+  connect-e2e:
+    name: Rust behavioral e2e
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python 3.11
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+
+      - name: Set up JDK 17
+        uses: actions/setup-java@v4
+        with:
+          distribution: temurin
+          java-version: "17"
+
+      - name: Install Rust toolchain
+        run: |
+          rustup toolchain install stable --profile minimal
+          rustup default stable
+
+      - name: Cache cargo & target
+        uses: actions/cache@v4
+        with:
+          path: |
+            ~/.cargo/registry
+            ~/.cargo/git
+            target
+          key: ${{ runner.os }}-cargo-e2e-${{ hashFiles('**/Cargo.lock') }}
+          restore-keys: ${{ runner.os }}-cargo-e2e-
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y protobuf-compiler
+
+      - name: Install Python dependencies (server-side UDF workers)
+        run: |
+          python -m pip install --upgrade pip
+          python -m pip install pandas pyarrow numpy
+
+      - name: Cache Spark ${{ env.SPARK_VERSION }} distribution
+        uses: actions/cache@v4
+        with:
+          path: ~/spark-dist
+          key: spark-dist-${{ env.SPARK_VERSION }}-hadoop3
+
+      - name: Download Apache Spark ${{ env.SPARK_VERSION }} (Connect server)
+        id: spark-download
+        run: |
+          PKG="spark-${SPARK_VERSION}-bin-hadoop3"
+          if [ ! -d "${HOME}/spark-dist/${PKG}" ]; then
+            URL="https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/${PKG}.tgz"
+            mkdir -p "${HOME}/spark-dist"
+            echo "Downloading ${URL}"
+            wget -q -O "${HOME}/spark-dist/${PKG}.tgz" "${URL}"
+            tar -xzf "${HOME}/spark-dist/${PKG}.tgz" -C "${HOME}/spark-dist"
+            rm -f "${HOME}/spark-dist/${PKG}.tgz"
+          else
+            echo "Using cached Spark distribution."
+          fi
+          echo "SPARK_HOME=${HOME}/spark-dist/${PKG}" >> "$GITHUB_OUTPUT"
+
+      - name: Start Spark Connect server
+        env:
+          PYSPARK_PYTHON: python
+          PYSPARK_DRIVER_PYTHON: python
+        run: |
+          SPARK_HOME="${{ steps.spark-download.outputs.SPARK_HOME }}"
+          "${SPARK_HOME}/sbin/start-connect-server.sh" \
+            --packages "org.apache.spark:spark-connect_2.13:${SPARK_VERSION}" \
+            --driver-memory 3g \
+            --conf spark.sql.session.localRelationSizeLimit=1073741824 \
+            > /tmp/connect-server.log 2>&1 &
+          for i in $(seq 1 90); do
+            if (echo > /dev/tcp/localhost/15002) 2>/dev/null; then
+              echo "Connect server is up (after ${i}s)."; break
+            fi
+            if [ "$i" -eq 90 ]; then echo "Server failed to start:"; cat /tmp/connect-server.log; exit 1; fi
+            sleep 1
+          done
+
+      - name: Rust behavioral e2e tests (against the live server)
+        env:
+          SPARK_REMOTE: sc://localhost:15002
+        run: |
+          cargo test -p apache-spark-connect \
+            --test e2e_functions --test e2e_columns --test e2e_dataframe \
+            --test e2e_coverage --test e2e_integration \
+            -- --test-threads=1
+
   # ---------------------------------------------------------------------------
   # E2E performance benchmark: our client vs reference, same 4.2.0 server.
   # ---------------------------------------------------------------------------

crates/spark-connect/src/conf.rs

@@ -124,7 +124,6 @@ impl Clone for RuntimeConf {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use crate::session::SparkSession;
 
     fn session() -> SparkSession {

crates/spark-connect/src/dataframe.rs

@@ -2431,7 +2431,7 @@ mod cache_tests {
         // The LocalRowIterator and collect() use the same underlying ExecutePlan.
         // The difference is purely in client-side consumption: streaming vs buffering.
         // We verify that to_local_iterator() creates the same iterator type.
-        let iter: LocalRowIterator;
+        let _iter: LocalRowIterator;
         // This test just verifies the type exists and is constructible.
         // A true integration test would create an actual stream.
     }

crates/spark-connect/src/merge.rs

@@ -220,7 +220,6 @@ impl WhenNotMatchedBySource {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
     use crate::session::SparkSession;
 
     fn session() -> SparkSession {

crates/spark-connect/src/ml.rs

@@ -12,7 +12,6 @@ use uuid::Uuid;
 
 use crate::dataframe::DataFrame;
 use crate::plan::LogicalPlan;
-use crate::session::SparkSession;
 
 /// ML parameter handling: stores name -> literal value mappings.
 ///
@@ -418,8 +417,6 @@ impl Estimator for StandardScaler {
                 OperatorType::Model,
             ),
             params: self.params.clone(),
-            input_col: self.input_col.clone(),
-            output_col: self.output_col.clone(),
         };
         Ok(Box::new(model))
     }
@@ -430,8 +427,6 @@ impl Estimator for StandardScaler {
 pub struct StandardScalerModel {
     operator: MlOperator,
     params: Params,
-    input_col: String,
-    output_col: String,
 }
 
 impl Transformer for StandardScalerModel {
@@ -1408,6 +1403,8 @@ impl Model for CrossValidatorModel {
 #[cfg(test)]
 mod tests {
     use super::*;
+    // Used only by the tests below (the non-test build does not reference it).
+    use crate::session::SparkSession;
 
     #[test]
     fn test_params_creation() {

crates/spark-connect/src/row.rs

@@ -291,7 +291,7 @@ mod tests {
 
         #[allow(clippy::approx_constant)]
         let d = Value::Double(3.14);
-        #[allow(clippy::approx_constant)]
+
         assert_eq!(d.as_f64(), Some(3.14));
 
         let s = Value::String("test".to_string());

crates/spark-connect/src/session.rs

@@ -552,7 +552,6 @@ impl Default for SparkSessionBuilder {
 /// Convert Arrow RecordBatch data to IPC bytes.
 /// Used by createDataFrame to package local data.
 fn rows_to_arrow_ipc(rows: &[Row], schema: &DataType) -> Result<Vec<u8>> {
-    use arrow::array::*;
     use arrow::datatypes::Schema as ArrowSchema;
     use arrow::ipc::writer::StreamWriter;
     use arrow::record_batch::RecordBatch;

crates/spark-connect/tests/catalog_golden.rs

@@ -13,7 +13,6 @@ use std::io::{BufRead, BufReader};
 use base64::{engine::general_purpose::STANDARD, Engine};
 use prost::Message;
 
-use spark_connect::catalog::Catalog;
 use spark_connect_proto as proto;
 
 /// Recursively clear fields that vary run-to-run and are not client-authored:
@@ -36,13 +35,10 @@ fn normalize_expression(e: &mut proto::Expression) {
     if let Some(expr_type) = &mut e.expr_type {
         // Recursively normalize nested expressions as needed
         use proto::expression::ExprType;
-        match expr_type {
-            ExprType::Alias(alias) => {
-                if let Some(expr) = &mut alias.expr {
-                    normalize_expression(expr);
-                }
+        if let ExprType::Alias(alias) = expr_type {
+            if let Some(expr) = &mut alias.expr {
+                normalize_expression(expr);
             }
-            _ => {}
         }
     }
 }

crates/spark-connect/tests/e2e_integration.rs

@@ -7,7 +7,6 @@
 //! Or simply skip with: `cargo test --test e2e_integration` (will skip all tests)
 
 use spark_connect::column::{col, lit};
-use spark_connect::functions;
 use spark_connect::row::Value;
 use spark_connect::session::SparkSession;
 use spark_connect::types::DataType;

crates/spark-connect/tests/ml_extra_golden.rs

@@ -90,7 +90,7 @@ fn test_vector_assembler_operator_name() {
 
 #[test]
 fn test_vector_assembler_proto() {
-    let mut assembler = VectorAssembler::new()
+    let assembler = VectorAssembler::new()
         .set_input_cols(vec!["col1", "col2"])
         .set_output_col("vector");
 

crates/spark-connect/tests/ml_golden.rs

@@ -24,10 +24,7 @@ fn normalize_expression(e: &mut proto::Expression) {
     use proto::expression::ExprType as T;
     e.common = None;
     if let Some(t) = e.expr_type.as_mut() {
-        match t {
-            T::Literal(_) => {}
-            _ => {}
-        }
+        if let T::Literal(_) = t {}
     }
 }
 

crates/spark-connect/tests/plans_golden.rs

@@ -25,7 +25,7 @@ fn normalize_relation(r: &mut proto::Relation) {
     if let Some(rel_type) = &mut r.rel_type {
         use proto::relation::RelType;
         match rel_type {
-            RelType::Range(range) => {}
+            RelType::Range(_range) => {}
             RelType::Sql(_) => {}
             RelType::Project(proj) => {
                 if let Some(input) = &mut proj.input {

crates/spark-connect/tests/readwriter_golden.rs

@@ -14,11 +14,8 @@ fn normalize_relation(r: &mut proto::Relation) {
     r.common = None;
     if let Some(_rel_type) = &mut r.rel_type {
         use proto::relation::RelType;
-        match _rel_type {
-            RelType::Read(_read) => {
-                // No recursive normalization needed for Read
-            }
-            _ => {}
+        if let RelType::Read(_read) = _rel_type {
+            // No recursive normalization needed for Read
         }
     }
 }

crates/spark-connect/tests/test_conf_tvf.rs

@@ -5,8 +5,6 @@
 #[cfg(test)]
 mod tests {
     use spark_connect::plan::LogicalPlan;
-    use spark_connect::table_arg::TableArg;
-    use spark_connect::tvf::TableValuedFunction;
 
     #[test]
     fn test_unresolved_table_valued_function_plan() {

crates/spark-connect/tests/window_golden.rs

@@ -10,10 +10,10 @@ use std::io::{BufRead, BufReader};
 
 use base64::{engine::general_purpose::STANDARD, Engine};
 use prost::Message;
-use spark_connect::column::{col, lit};
+use spark_connect::column::col;
 use spark_connect::expression::{Expression, LiteralExpression};
 use spark_connect::functions;
-use spark_connect::window::{FrameBound, FrameType, Window, WindowSpec};
+use spark_connect::window::{FrameBound, WindowSpec};
 use spark_connect_proto as proto;
 
 /// An integer literal Column (pyspark infers 32-bit `integer` for small ints).
@@ -261,7 +261,7 @@ fn test_window_golden() {
     let mut matched = 0;
 
     for (name, expected) in &goldens {
-        let built = build(name).expect(&format!("missing build case for {}", name));
+        let built = build(name).unwrap_or_else(|| panic!("missing build case for {}", name));
         let mut actual = built.to_proto();
         normalize(&mut actual);