This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark-connect-rust.git
The following commit(s) were added to refs/heads/master by this push:
new 66c2af8 [SPARK-59032][INFRA] Run behavioral e2e in a separate job so
the parity gate gets a clean Connect server
66c2af8 is described below
commit 66c2af8245ca3401a1f95209ff5ac260135d3261
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Fri Aug 28 14:36:09 2026 +0900
[SPARK-59032][INFRA] Run behavioral e2e in a separate job so the parity
gate gets a clean Connect server
### What changes were proposed in this pull request?
**Bandaid for the arrow-UDF parity hang (SPARK-59032).** Move the Rust
behavioral
e2e tests out of the parity job into their own `connect-e2e` job with its
own Connect
server, so the parity gate starts against a server nothing else has touched.
**Root cause (why this helps):** the parity gate hung on
`test_parity_arrow_python_udf.py` only when the behavioral e2e cargo tests
had already
run against the *same* shared Connect server. `test_python_udf_segfault`
deliberately
crashes a Python worker (`ctypes.string_at(0)`, faulthandler enabled); on a
server the
e2e tests had already stressed, that left the server's Python-worker reuse
wedged, so a
later UDF query (`test_raise_stop_iteration`'s `range(5)`) never produced
results - the
server streamed only `ExecutionProgress` responses (0 Arrow batches, no
`ResultComplete`),
and the reattachable client dutifully re-attached forever. Diagnosed from a
per-test
client thread dump + server jstack + a response-type histogram of the
stalled streams.
Giving the parity run a clean server sidesteps the wedge. **No test is
skipped.** This is
an infra-level mitigation; a proper client/server worker-recovery fix is a
follow-up.
Also piggybacked here (unrelated cleanup, from #63):
- Remove compiler-flagged unused Rust imports and redundant
`StandardScalerModel`
fields, plus unused imports/vars in tests (a `clippy --all-targets`
sweep). No
behavior change.
### Why are the changes needed?
The parity gate timed out on every run because of the shared-server
interaction above.
This makes it green without skipping any test.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
CI: the parity gate runs against a fresh server; the behavioral e2e tests
run in their
own `connect-e2e` job. `cargo build --release` is clean and the
`apache-spark-connect`
unit tests pass.
This pull request and its description were written by Isaac.
Closes #58 from HyukjinKwon/fix-arrow-parity-hang.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.github/workflows/build_python_connect.yml | 118 +++++++++++++++++++++---
crates/spark-connect/src/conf.rs | 1 -
crates/spark-connect/src/dataframe.rs | 2 +-
crates/spark-connect/src/merge.rs | 1 -
crates/spark-connect/src/ml.rs | 7 +-
crates/spark-connect/src/row.rs | 2 +-
crates/spark-connect/src/session.rs | 1 -
crates/spark-connect/tests/catalog_golden.rs | 10 +-
crates/spark-connect/tests/e2e_integration.rs | 1 -
crates/spark-connect/tests/ml_extra_golden.rs | 2 +-
crates/spark-connect/tests/ml_golden.rs | 5 +-
crates/spark-connect/tests/plans_golden.rs | 2 +-
crates/spark-connect/tests/readwriter_golden.rs | 7 +-
crates/spark-connect/tests/test_conf_tvf.rs | 2 -
crates/spark-connect/tests/window_golden.rs | 6 +-
15 files changed, 120 insertions(+), 47 deletions(-)
diff --git a/.github/workflows/build_python_connect.yml
b/.github/workflows/build_python_connect.yml
index 2896229..7117ba7 100644
--- a/.github/workflows/build_python_connect.yml
+++ b/.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.
# ---------------------------------------------------------------------------
diff --git a/crates/spark-connect/src/conf.rs b/crates/spark-connect/src/conf.rs
index d6454d5..c72401f 100644
--- a/crates/spark-connect/src/conf.rs
+++ b/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 {
diff --git a/crates/spark-connect/src/dataframe.rs
b/crates/spark-connect/src/dataframe.rs
index 718d045..3f43030 100644
--- a/crates/spark-connect/src/dataframe.rs
+++ b/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.
}
diff --git a/crates/spark-connect/src/merge.rs
b/crates/spark-connect/src/merge.rs
index 3f03314..7cbe120 100644
--- a/crates/spark-connect/src/merge.rs
+++ b/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 {
diff --git a/crates/spark-connect/src/ml.rs b/crates/spark-connect/src/ml.rs
index 892d94b..a0dd840 100644
--- a/crates/spark-connect/src/ml.rs
+++ b/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() {
diff --git a/crates/spark-connect/src/row.rs b/crates/spark-connect/src/row.rs
index 362d6e1..7a16d21 100644
--- a/crates/spark-connect/src/row.rs
+++ b/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());
diff --git a/crates/spark-connect/src/session.rs
b/crates/spark-connect/src/session.rs
index b8c6a79..b40ff88 100644
--- a/crates/spark-connect/src/session.rs
+++ b/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;
diff --git a/crates/spark-connect/tests/catalog_golden.rs
b/crates/spark-connect/tests/catalog_golden.rs
index 41b2bb7..460767a 100644
--- a/crates/spark-connect/tests/catalog_golden.rs
+++ b/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);
}
- _ => {}
}
}
}
diff --git a/crates/spark-connect/tests/e2e_integration.rs
b/crates/spark-connect/tests/e2e_integration.rs
index 43910e8..f72bfa8 100644
--- a/crates/spark-connect/tests/e2e_integration.rs
+++ b/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;
diff --git a/crates/spark-connect/tests/ml_extra_golden.rs
b/crates/spark-connect/tests/ml_extra_golden.rs
index 7bb7ff2..2456a7a 100644
--- a/crates/spark-connect/tests/ml_extra_golden.rs
+++ b/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");
diff --git a/crates/spark-connect/tests/ml_golden.rs
b/crates/spark-connect/tests/ml_golden.rs
index 037bcd2..10f942a 100644
--- a/crates/spark-connect/tests/ml_golden.rs
+++ b/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 {}
}
}
diff --git a/crates/spark-connect/tests/plans_golden.rs
b/crates/spark-connect/tests/plans_golden.rs
index 66d16a7..14d4cb1 100644
--- a/crates/spark-connect/tests/plans_golden.rs
+++ b/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 {
diff --git a/crates/spark-connect/tests/readwriter_golden.rs
b/crates/spark-connect/tests/readwriter_golden.rs
index 95da505..de182b7 100644
--- a/crates/spark-connect/tests/readwriter_golden.rs
+++ b/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
}
}
}
diff --git a/crates/spark-connect/tests/test_conf_tvf.rs
b/crates/spark-connect/tests/test_conf_tvf.rs
index 3bf5454..da87f69 100644
--- a/crates/spark-connect/tests/test_conf_tvf.rs
+++ b/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() {
diff --git a/crates/spark-connect/tests/window_golden.rs
b/crates/spark-connect/tests/window_golden.rs
index 3c55331..5528d5e 100644
--- a/crates/spark-connect/tests/window_golden.rs
+++ b/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);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
