You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/08/29 11:04:59 UTC
(spark-connect-rust) branch branch-4.2 updated: [SPARK-59109] Fix PySpark API-surface parity: properties, show() default, kwarg names, __version__
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark-connect-rust.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 5b745fe [SPARK-59109] Fix PySpark API-surface parity: properties, show() default, kwarg names, __version__
5b745fe is described below
commit 5b745fe65ddab3e2a1f1cd56885630e2d121f11f
Author: Hyukjin Kwon <hy...@databricks.com>
AuthorDate: Sat Aug 29 19:57:43 2026 +0900
[SPARK-59109] Fix PySpark API-surface parity: properties, show() default, kwarg names, __version__
An audit of the Rust-backed drop-in against real PySpark 4.2.0 (which the
transport-level official-suite gate never exercises) found basic API-surface
gaps. Fix them and add offline guards.
- pyspark.__version__ was "0.1.0"; source it from pyspark.version ("4.2.0").
- Expose version (SparkSession) and dtypes / executionInfo (DataFrame) as
properties (#[getter]) instead of methods, matching PySpark attribute access.
- DataFrame.show now defaults to (n=20, truncate=True, vertical=False) so
df.show() works; truncate/vertical are accepted for signature parity.
- Rename PyO3 method parameters from snake_case to PySpark's camelCase so
keyword calls match: SparkSession.sql(sqlQuery), table(tableName),
interruptOperation(op_id), removeProgressHandler(handler), addArtifact(*path,
pyfile,archive,file); DataFrame.limit/take(num), offset(num),
coalesce/repartition(numPartitions), colRegex(colName), withColumn(colName),
withColumns(*colsMap), withColumnsRenamed(colsMap), alias(alias),
groupingSets(groupingSets), approxQuantile(relativeError), writeTo(table),
foreach/foreachPartition(f), metadataColumn(colName),
nearestByJoin(rankingExpression,numResults,joinType); Column.cast/astype/
try_cast(dataType), substr(startPos), like/rlike/ilike(other), over(window),
between(lowerBound,upperBound), withField(fieldName,col), and make
Column.alias/name variadic (*alias) to match PySpark.
Add offline introspection tests (no server) asserting __version__, the three
properties, show()'s defaults, and the corrected keyword-parameter names.
Verified: the API/signature audit reports zero property-vs-method and zero
required-arg mismatches vs PySpark 4.2.0, the 229-test offline suite passes, and
a live run against a Spark Connect server confirms df.show(), spark.version,
df.dtypes, and keyword calls (limit(num=), cast(dataType=), sql(sqlQuery=), ...).
Co-authored-by: Isaac <no...@databricks.com>
---
crates/pyspark-rs/src/column.rs | 71 +++++++++++++-----------
crates/pyspark-rs/src/dataframe.rs | 104 +++++++++++++++++++-----------------
crates/pyspark-rs/src/session.rs | 34 ++++++++----
python/pyspark/__init__.py | 2 +-
python/tests/test_dropin_offline.py | 55 +++++++++++++++++++
5 files changed, 176 insertions(+), 90 deletions(-)
diff --git a/crates/pyspark-rs/src/column.rs b/crates/pyspark-rs/src/column.rs
index fea66d0..a332df7 100644
--- a/crates/pyspark-rs/src/column.rs
+++ b/crates/pyspark-rs/src/column.rs
@@ -50,24 +50,35 @@ fn py_obj_to_column(obj: &Bound<'_, PyAny>) -> PyResult<Column> {
impl PyColumn {
/// Alias the column, optionally attaching column metadata (a dict), mirroring
/// `Column.alias(name, metadata=...)`.
- #[pyo3(signature = (name, metadata=None))]
+ #[pyo3(signature = (*alias, metadata=None))]
fn alias(
&self,
- name: &str,
+ alias: Vec<String>,
metadata: Option<std::collections::HashMap<String, String>>,
) -> PyColumn {
+ // Mirrors Column.alias(*alias, **kwargs): the common single-name form aliases the
+ // column; extra names beyond the first (multi-output aliasing) are not applied here.
+ let name = match alias.first() {
+ Some(n) => n.clone(),
+ None => return PyColumn::new(self.column.clone()),
+ };
match metadata {
- None => PyColumn::new(self.column.clone().alias(name)),
+ None => PyColumn::new(self.column.clone().alias(&name)),
Some(m) => {
let md: std::collections::BTreeMap<String, String> = m.into_iter().collect();
- PyColumn::new(self.column.clone().alias_with_metadata(name, md))
+ PyColumn::new(self.column.clone().alias_with_metadata(&name, md))
}
}
}
/// Alias for `alias` (pyspark `Column.name`).
- fn name(&self, name: &str) -> PyColumn {
- PyColumn::new(self.column.clone().name(name))
+ #[pyo3(signature = (*alias, metadata=None))]
+ fn name(
+ &self,
+ alias: Vec<String>,
+ metadata: Option<std::collections::HashMap<String, String>>,
+ ) -> PyColumn {
+ self.alias(alias, metadata)
}
/// Apply a transformation function to this column. Mirrors `Column.transform(f)`,
@@ -85,30 +96,30 @@ impl PyColumn {
/// Cast to a different type. Accepts a `DataType` or a DDL type string,
/// matching `pyspark.sql.Column.cast`.
- fn cast(&self, data_type: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
+ fn cast(&self, dataType: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
// A DDL string keeps the unparsed type_str form (reference cast("int"));
// any DataType object is converted to a typed cast.
- if let Ok(s) = data_type.extract::<String>() {
+ if let Ok(s) = dataType.extract::<String>() {
Ok(PyColumn::new(self.column.clone().cast_str(&s)))
} else {
- let dt = crate::types::py_to_data_type(data_type)?;
+ let dt = crate::types::py_to_data_type(dataType)?;
Ok(PyColumn::new(self.column.clone().cast(dt)))
}
}
/// Alias for `cast` (pyspark `Column.astype`).
- fn astype(&self, data_type: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
- self.cast(data_type)
+ fn astype(&self, dataType: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
+ self.cast(dataType)
}
/// Try to cast, yielding NULL on failure. Accepts a `DataType` or DDL string.
/// Mirrors `pyspark.sql.Column.try_cast`.
#[pyo3(name = "try_cast")]
- fn try_cast(&self, data_type: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
- if let Ok(s) = data_type.extract::<String>() {
+ fn try_cast(&self, dataType: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
+ if let Ok(s) = dataType.extract::<String>() {
Ok(PyColumn::new(self.column.clone().try_cast_str(&s)))
} else {
- let dt = crate::types::py_to_data_type(data_type)?;
+ let dt = crate::types::py_to_data_type(dataType)?;
Ok(PyColumn::new(self.column.clone().try_cast(dt)))
}
}
@@ -133,8 +144,8 @@ impl PyColumn {
}
/// Substring extraction (start and length as columns).
- fn substr(&self, start: &Bound<'_, PyAny>, length: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
- let start_col = py_obj_to_column(start)?;
+ fn substr(&self, startPos: &Bound<'_, PyAny>, length: &Bound<'_, PyAny>) -> PyResult<PyColumn> {
+ let start_col = py_obj_to_column(startPos)?;
let length_col = py_obj_to_column(length)?;
Ok(PyColumn::new(
self.column.clone().substr(start_col, length_col),
@@ -142,13 +153,13 @@ impl PyColumn {
}
/// Pattern matching with LIKE.
- fn like(&self, pattern: &str) -> PyColumn {
- PyColumn::new(self.column.clone().like(pattern))
+ fn like(&self, other: &str) -> PyColumn {
+ PyColumn::new(self.column.clone().like(other))
}
/// Pattern matching with RLIKE (regex).
- fn rlike(&self, pattern: &str) -> PyColumn {
- PyColumn::new(self.column.clone().rlike(pattern))
+ fn rlike(&self, other: &str) -> PyColumn {
+ PyColumn::new(self.column.clone().rlike(other))
}
/// Check if contains another column value.
@@ -333,8 +344,8 @@ impl PyColumn {
}
/// Apply a window function over a window specification.
- fn over(&self, window_spec: &crate::window::PyWindowSpec) -> PyColumn {
- PyColumn::new(self.column.clone().over(window_spec.spec.clone()))
+ fn over(&self, window: &crate::window::PyWindowSpec) -> PyColumn {
+ PyColumn::new(self.column.clone().over(window.spec.clone()))
}
/// String representation, mirroring `pyspark.sql.connect.column.Column.__repr__`
@@ -418,9 +429,9 @@ impl PyColumn {
}
// --- named Column methods (PySpark camelCase) ---
- fn between(&self, py: Python<'_>, lower: Py<PyAny>, upper: Py<PyAny>) -> PyResult<PyColumn> {
- let l = to_column(&lower.bind(py))?;
- let u = to_column(&upper.bind(py))?;
+ fn between(&self, py: Python<'_>, lowerBound: Py<PyAny>, upperBound: Py<PyAny>) -> PyResult<PyColumn> {
+ let l = to_column(&lowerBound.bind(py))?;
+ let u = to_column(&upperBound.bind(py))?;
Ok(PyColumn::new(self.column.clone().between(l, u)))
}
#[pyo3(name = "bitwiseAND")]
@@ -438,8 +449,8 @@ impl PyColumn {
let o = to_column(&other.bind(py))?;
Ok(PyColumn::new(self.column.clone().bitwise_xor(o)))
}
- fn ilike(&self, pattern: &str) -> PyColumn {
- PyColumn::new(self.column.clone().ilike(pattern))
+ fn ilike(&self, other: &str) -> PyColumn {
+ PyColumn::new(self.column.clone().ilike(other))
}
#[pyo3(name = "isNaN")]
fn is_nan(&self) -> PyColumn {
@@ -475,9 +486,9 @@ impl PyColumn {
Ok(PyColumn::new(self.column.clone().isin(vals)))
}
#[pyo3(name = "withField")]
- fn with_field(&self, py: Python<'_>, field_name: &str, value: Py<PyAny>) -> PyResult<PyColumn> {
- let v = to_column(&value.bind(py))?;
- Ok(PyColumn::new(self.column.clone().with_field(field_name, v)))
+ fn with_field(&self, py: Python<'_>, fieldName: &str, col: Py<PyAny>) -> PyResult<PyColumn> {
+ let v = to_column(&col.bind(py))?;
+ Ok(PyColumn::new(self.column.clone().with_field(fieldName, v)))
}
#[pyo3(name = "dropFields", signature = (*field_names))]
fn drop_fields(&self, field_names: Vec<String>) -> PyColumn {
diff --git a/crates/pyspark-rs/src/dataframe.rs b/crates/pyspark-rs/src/dataframe.rs
index 6144158..48bd5b1 100644
--- a/crates/pyspark-rs/src/dataframe.rs
+++ b/crates/pyspark-rs/src/dataframe.rs
@@ -151,8 +151,8 @@ impl PyDataFrame {
}
/// Add or replace a column.
- fn withColumn(&self, name: &str, col: &PyColumn) -> PyDataFrame {
- PyDataFrame::new(self.dataframe.with_column(name, col.column.clone()))
+ fn withColumn(&self, colName: &str, col: &PyColumn) -> PyDataFrame {
+ PyDataFrame::new(self.dataframe.with_column(colName, col.column.clone()))
}
/// Rename a column.
@@ -161,13 +161,15 @@ impl PyDataFrame {
}
/// Add or replace multiple columns from a {name: Column} mapping.
- #[pyo3(name = "withColumns")]
- fn with_columns(&self, cols: &Bound<'_, PyDict>) -> PyResult<PyDataFrame> {
- let mut pairs = Vec::with_capacity(cols.len());
- for (k, v) in cols.iter() {
- let name: String = k.extract()?;
- let pycol = v.extract::<PyRef<PyColumn>>()?;
- pairs.push((name, pycol.column.clone()));
+ #[pyo3(name = "withColumns", signature = (*colsMap))]
+ fn with_columns(&self, colsMap: Vec<Bound<'_, PyDict>>) -> PyResult<PyDataFrame> {
+ let mut pairs = Vec::new();
+ for d in &colsMap {
+ for (k, v) in d.iter() {
+ let name: String = k.extract()?;
+ let pycol = v.extract::<PyRef<PyColumn>>()?;
+ pairs.push((name, pycol.column.clone()));
+ }
}
Ok(PyDataFrame::new(self.dataframe.with_columns(pairs)))
}
@@ -193,13 +195,13 @@ impl PyDataFrame {
}
/// Limit the number of rows.
- fn limit(&self, n: i32) -> PyDataFrame {
- PyDataFrame::new(self.dataframe.limit(n))
+ fn limit(&self, num: i32) -> PyDataFrame {
+ PyDataFrame::new(self.dataframe.limit(num))
}
/// Skip the first n rows.
- fn offset(&self, n: i32) -> PyDataFrame {
- PyDataFrame::new(self.dataframe.offset(n))
+ fn offset(&self, num: i32) -> PyDataFrame {
+ PyDataFrame::new(self.dataframe.offset(num))
}
/// Remove duplicate rows.
@@ -395,16 +397,16 @@ impl PyDataFrame {
/// first argument may instead be a Column/str - `df.repartition("country")` /
/// `df.repartition(col("country"))` repartition by that column at the default
/// partition count (no explicit number).
- #[pyo3(signature = (num_partitions, *cols))]
+ #[pyo3(signature = (numPartitions, *cols))]
fn repartition(
&self,
py: Python<'_>,
- num_partitions: Bound<'_, PyAny>,
+ numPartitions: Bound<'_, PyAny>,
cols: Vec<Bound<'_, PyAny>>,
) -> PyResult<PyDataFrame> {
// First arg is an int -> a partition count; otherwise it's a partition column
// (str/Column) and the count defaults to the server's (passed as 0 = unset).
- if let Ok(n) = num_partitions.extract::<i32>() {
+ if let Ok(n) = numPartitions.extract::<i32>() {
if cols.is_empty() {
return Ok(PyDataFrame::new(self.dataframe.repartition(n)));
}
@@ -417,7 +419,7 @@ impl PyDataFrame {
))
} else {
// Column-first form: the first arg is itself a partition column.
- let mut all = vec![num_partitions];
+ let mut all = vec![numPartitions];
all.extend(cols);
let _ = py;
let exprs: Vec<_> = to_column_list(all)?
@@ -431,8 +433,8 @@ impl PyDataFrame {
}
/// Coalesce.
- fn coalesce(&self, num_partitions: i32) -> PyDataFrame {
- PyDataFrame::new(self.dataframe.coalesce(num_partitions))
+ fn coalesce(&self, numPartitions: i32) -> PyDataFrame {
+ PyDataFrame::new(self.dataframe.coalesce(numPartitions))
}
/// Convert to DataFrame with new column names.
@@ -443,8 +445,8 @@ impl PyDataFrame {
}
/// Alias this DataFrame.
- fn alias(&self, name: &str) -> PyDataFrame {
- PyDataFrame::new(self.dataframe.alias(name))
+ fn alias(&self, alias: &str) -> PyDataFrame {
+ PyDataFrame::new(self.dataframe.alias(alias))
}
/// Group by columns for aggregation.
@@ -472,9 +474,9 @@ impl PyDataFrame {
/// Grouping sets aggregation: `groupingSets([[a, b], [a], []])`.
#[pyo3(name = "groupingSets")]
- fn grouping_sets(&self, sets: Vec<Vec<Bound<'_, PyAny>>>) -> PyResult<PyGroupedData> {
- let mut out: Vec<Vec<spark_connect::column::Column>> = Vec::with_capacity(sets.len());
- for s in sets {
+ fn grouping_sets(&self, groupingSets: Vec<Vec<Bound<'_, PyAny>>>) -> PyResult<PyGroupedData> {
+ let mut out: Vec<Vec<spark_connect::column::Column>> = Vec::with_capacity(groupingSets.len());
+ for s in groupingSets {
out.push(to_column_list(s)?);
}
Ok(PyGroupedData::new(self.dataframe.grouping_sets(out)))
@@ -539,18 +541,18 @@ impl PyDataFrame {
#[pyo3(name = "withColumnsRenamed")]
fn with_columns_renamed(
&self,
- renames: std::collections::HashMap<String, String>,
+ colsMap: std::collections::HashMap<String, String>,
) -> PyDataFrame {
PyDataFrame::new(
self.dataframe
- .with_columns_renamed(renames.into_iter().collect()),
+ .with_columns_renamed(colsMap.into_iter().collect()),
)
}
/// Select columns whose names match a regex.
#[pyo3(name = "colRegex")]
- fn col_regex(&self, col_name: &str) -> PyDataFrame {
- PyDataFrame::new(self.dataframe.col_regex(col_name))
+ fn col_regex(&self, colName: &str) -> PyDataFrame {
+ PyDataFrame::new(self.dataframe.col_regex(colName))
}
/// Basic statistics for the given columns (count/mean/stddev/min/max).
@@ -685,13 +687,13 @@ impl PyDataFrame {
py: Python<'_>,
col: &str,
probabilities: Vec<f64>,
- relative_error: f64,
+ relativeError: f64,
) -> PyResult<Vec<f64>> {
use spark_connect::row::Value;
let df = self
.dataframe
.stat()
- .approx_quantile(vec![col], probabilities, relative_error);
+ .approx_quantile(vec![col], probabilities, relativeError);
let rows = py.detach(|| df.collect()).to_pyerr()?;
let quantiles = match rows.first().and_then(|r| r.get(0)) {
Some(Value::List(outer)) => match outer.first() {
@@ -1132,11 +1134,16 @@ impl PyDataFrame {
}
/// Show the first n rows.
- fn show(&self, py: Python<'_>, n: usize) -> PyResult<()> {
+ #[pyo3(signature = (n=20, truncate=true, vertical=false))]
+ fn show(&self, py: Python<'_>, n: usize, truncate: bool, vertical: bool) -> PyResult<()> {
+ // truncate/vertical are accepted for PySpark signature parity; the Rust core
+ // currently renders with its default formatting.
+ let _ = (truncate, vertical);
py.detach(|| self.dataframe.show(n)).to_pyerr()
}
/// Execution metrics from the most recent action. Mirrors `DataFrame.executionInfo`.
+ #[getter]
#[pyo3(name = "executionInfo")]
fn execution_info(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
let info = self.dataframe.execution_info().to_pyerr()?;
@@ -1213,8 +1220,8 @@ impl PyDataFrame {
}
/// Get the first n rows.
- fn take(&self, py: Python<'_>, n: usize) -> PyResult<Vec<PyRow>> {
- let rows = py.detach(|| self.dataframe.take(n)).to_pyerr()?;
+ fn take(&self, py: Python<'_>, num: usize) -> PyResult<Vec<PyRow>> {
+ let rows = py.detach(|| self.dataframe.take(num)).to_pyerr()?;
Ok(rows.into_iter().map(PyRow::new).collect())
}
@@ -1287,6 +1294,7 @@ impl PyDataFrame {
}
/// Get the schema as a list of (name, type) tuples.
+ #[getter]
fn dtypes(&self) -> PyResult<Vec<(String, String)>> {
let schema = self.dataframe.schema().to_pyerr()?;
match schema {
@@ -1318,9 +1326,9 @@ impl PyDataFrame {
/// The V2 writer interface (`df.writeTo("table")`).
#[pyo3(name = "writeTo")]
- fn write_to(&self, table_name: &str) -> PyDataFrameWriterV2 {
+ fn write_to(&self, table: &str) -> PyDataFrameWriterV2 {
PyDataFrameWriterV2 {
- inner: Some(self.dataframe.write_to(table_name)),
+ inner: Some(self.dataframe.write_to(table)),
}
}
@@ -1407,43 +1415,43 @@ impl PyDataFrame {
/// `DataFrame.foreach` — run a function per row for side effects.
#[pyo3(name = "foreach")]
- fn foreach(&self, py: Python<'_>, func: Bound<'_, PyAny>) -> PyResult<()> {
- let udf = build_side_effect_udf(py, "foreach", &func)?;
+ fn foreach(&self, py: Python<'_>, f: Bound<'_, PyAny>) -> PyResult<()> {
+ let udf = build_side_effect_udf(py, "foreach", &f)?;
self.dataframe.foreach(udf).to_pyerr()
}
/// `DataFrame.foreachPartition` — run a function per partition for side effects.
#[pyo3(name = "foreachPartition")]
- fn foreach_partition(&self, py: Python<'_>, func: Bound<'_, PyAny>) -> PyResult<()> {
- let udf = build_side_effect_udf(py, "foreachPartition", &func)?;
+ fn foreach_partition(&self, py: Python<'_>, f: Bound<'_, PyAny>) -> PyResult<()> {
+ let udf = build_side_effect_udf(py, "foreachPartition", &f)?;
self.dataframe.foreach_partition(udf).to_pyerr()
}
/// `DataFrame.nearestByJoin`.
- #[pyo3(name = "nearestByJoin", signature = (other, ranking_expression, num_results = 1, mode = "brute", direction = "asc", join_type = "inner"))]
+ #[pyo3(name = "nearestByJoin", signature = (other, rankingExpression, numResults = 1, mode = "brute", direction = "asc", joinType = "inner"))]
fn nearest_by_join(
&self,
other: &PyDataFrame,
- ranking_expression: &PyColumn,
- num_results: i32,
+ rankingExpression: &PyColumn,
+ numResults: i32,
mode: &str,
direction: &str,
- join_type: &str,
+ joinType: &str,
) -> PyDataFrame {
PyDataFrame::new(self.dataframe.nearest_by_join(
&other.dataframe,
- ranking_expression.column.clone(),
- num_results,
+ rankingExpression.column.clone(),
+ numResults,
mode,
direction,
- join_type,
+ joinType,
))
}
/// `DataFrame.metadataColumn`.
#[pyo3(name = "metadataColumn")]
- fn metadata_column(&self, name: &str) -> PyColumn {
- PyColumn::new(self.dataframe.metadata_column(name))
+ fn metadata_column(&self, colName: &str) -> PyColumn {
+ PyColumn::new(self.dataframe.metadata_column(colName))
}
}
diff --git a/crates/pyspark-rs/src/session.rs b/crates/pyspark-rs/src/session.rs
index 45b59c7..ea91b6b 100644
--- a/crates/pyspark-rs/src/session.rs
+++ b/crates/pyspark-rs/src/session.rs
@@ -244,8 +244,8 @@ impl PySparkSession {
}
/// Execute a SQL query and return a DataFrame.
- fn sql(&self, query: &str) -> PyResult<PyDataFrame> {
- let df = self.session.sql(query).to_pyerr()?;
+ fn sql(&self, sqlQuery: &str) -> PyResult<PyDataFrame> {
+ let df = self.session.sql(sqlQuery).to_pyerr()?;
Ok(PyDataFrame::new(df))
}
@@ -592,8 +592,8 @@ impl PySparkSession {
/// Remove a progress handler by id. Mirrors `removeProgressHandler`.
#[pyo3(name = "removeProgressHandler")]
- fn remove_progress_handler(&self, id: u64) {
- self.session.remove_progress_handler(id);
+ fn remove_progress_handler(&self, handler: u64) {
+ self.session.remove_progress_handler(handler);
}
/// Remove all progress handlers. Mirrors `clearProgressHandlers`.
@@ -628,12 +628,13 @@ impl PySparkSession {
.to_pyerr()
}
+ #[getter]
fn version(&self, py: Python<'_>) -> PyResult<String> {
py.detach(|| self.session.version()).to_pyerr()
}
- fn table(&self, table_name: &str) -> PyResult<PyDataFrame> {
- Ok(PyDataFrame::new(self.session.table(table_name).to_pyerr()?))
+ fn table(&self, tableName: &str) -> PyResult<PyDataFrame> {
+ Ok(PyDataFrame::new(self.session.table(tableName).to_pyerr()?))
}
#[pyo3(name = "emptyDataFrame")]
@@ -654,8 +655,8 @@ impl PySparkSession {
}
#[pyo3(name = "interruptOperation")]
- fn interrupt_operation(&self, py: Python<'_>, operation_id: &str) -> PyResult<Vec<String>> {
- py.detach(|| self.session.interrupt_operation(operation_id))
+ fn interrupt_operation(&self, py: Python<'_>, op_id: &str) -> PyResult<Vec<String>> {
+ py.detach(|| self.session.interrupt_operation(op_id))
.to_pyerr()
}
@@ -686,9 +687,20 @@ impl PySparkSession {
py.detach(|| self.session.add_artifacts(&refs)).to_pyerr()
}
- #[pyo3(name = "addArtifact")]
- fn add_artifact(&self, py: Python<'_>, path: &str) -> PyResult<()> {
- py.detach(|| self.session.add_artifact(path)).to_pyerr()
+ #[pyo3(name = "addArtifact", signature = (*path, pyfile=false, archive=false, file=false))]
+ fn add_artifact(
+ &self,
+ py: Python<'_>,
+ path: Vec<String>,
+ pyfile: bool,
+ archive: bool,
+ file: bool,
+ ) -> PyResult<()> {
+ // pyfile/archive/file are accepted for PySpark signature parity; artifact
+ // classification is handled by the Rust transport / server.
+ let _ = (pyfile, archive, file);
+ let refs: Vec<&str> = path.iter().map(|s| s.as_str()).collect();
+ py.detach(|| self.session.add_artifacts(&refs)).to_pyerr()
}
#[pyo3(name = "newSession")]
diff --git a/python/pyspark/__init__.py b/python/pyspark/__init__.py
index a27f047..6706a73 100644
--- a/python/pyspark/__init__.py
+++ b/python/pyspark/__init__.py
@@ -2,7 +2,7 @@
#
# Exposes the Rust-backed Spark Connect client as a pure-Python package.
-__version__ = "0.1.0"
+from pyspark.version import __version__ # noqa: F401 (tracks the Spark version we target)
from typing import Callable, TypeVar, Union
diff --git a/python/tests/test_dropin_offline.py b/python/tests/test_dropin_offline.py
index 9856d17..fde19af 100644
--- a/python/tests/test_dropin_offline.py
+++ b/python/tests/test_dropin_offline.py
@@ -2394,3 +2394,58 @@ def test_drop_metadata_and_create_row():
r2 = _create_row(r, [3, 4])
assert r2["a"] == 3 and r2["b"] == 4
assert list(r.__fields__) == ["a", "b"]
+
+
+# ------------------------------------------- API parity: version, properties, signatures
+#
+# These guard the PySpark-surface parity that the transport-level official-suite gate does
+# not exercise: the reported version, property-vs-method access, show()'s default, and the
+# public keyword-argument names (camelCase, not the Rust snake_case).
+
+import inspect # noqa: E402
+
+import pyspark # noqa: E402
+from pyspark.sql import SparkSession as _SS, DataFrame as _DF, Column as _Col # noqa: E402
+
+
+def test_dropin_version_is_spark_version():
+ assert pyspark.__version__ == "4.2.0"
+
+
+@pytest.mark.parametrize("cls,name", [(_SS, "version"), (_DF, "dtypes"), (_DF, "executionInfo")])
+def test_property_not_method(cls, name):
+ # PySpark accesses these without parentheses; the Rust getter is a getset_descriptor.
+ assert type(inspect.getattr_static(cls, name)).__name__ == "getset_descriptor"
+
+
+def test_show_has_no_required_args_and_defaults_n_to_20():
+ sig = inspect.signature(_DF.show)
+ required = [
+ p.name
+ for p in sig.parameters.values()
+ if p.name not in ("self", "cls")
+ and p.kind in (p.POSITIONAL_OR_KEYWORD, p.POSITIONAL_ONLY, p.KEYWORD_ONLY)
+ and p.default is inspect._empty
+ ]
+ assert required == [], required # df.show() must work
+ assert sig.parameters["n"].default == 20
+
+
+@pytest.mark.parametrize(
+ "cls,method,param",
+ [
+ (_SS, "sql", "sqlQuery"),
+ (_SS, "table", "tableName"),
+ (_DF, "limit", "num"),
+ (_DF, "take", "num"),
+ (_DF, "withColumn", "colName"),
+ (_DF, "repartition", "numPartitions"),
+ (_Col, "cast", "dataType"),
+ (_Col, "astype", "dataType"),
+ (_Col, "substr", "startPos"),
+ (_Col, "between", "lowerBound"),
+ (_Col, "over", "window"),
+ ],
+)
+def test_keyword_param_names_match_pyspark(cls, method, param):
+ assert param in inspect.signature(getattr(cls, method)).parameters
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org