From 09cf885edd8353c47d91a580782ee237513e9dd0 Mon Sep 17 00:00:00 2001 From: Hyukjin Kwon Date: Sun, 8 Nov 2026 19:11:16 +0900 Subject: [PATCH] [SPARK-61179][CONNECT][PYTHON] Unset a DataFrameReader option of None, and take the description of Catalog.createExternalTable, in the Python client as PySpark does DataFrameReader.option(key, None), DataFrameReader.options(key=None) and DataFrameReader.load(..., key=None) unset the option in PySpark, while the format methods (csv(path, header=None) and the like) skip None and keep it. The client skipped None in all of them, so an earlier value stayed. The reader now unsets the option for the first three. Catalog.createExternalTable passes its keyword arguments to createTable, so description is createTable's description, which becomes the table comment. The client sent it as a table option instead, so there was no comment and a "description" property. Co-authored-by: Isaac --- crates/pyspark-rs/src/catalog.rs | 7 ++-- crates/pyspark-rs/src/readwriter.rs | 56 ++++++++++++++------------ crates/spark-connect/src/readwriter.rs | 6 +++ python/tests/test_dropin_e2e.py | 36 +++++++++++++++++ 4 files changed, 77 insertions(+), 28 deletions(-) diff --git a/crates/pyspark-rs/src/catalog.rs b/crates/pyspark-rs/src/catalog.rs index 80680ba5..d0ca8da4 100644 --- a/crates/pyspark-rs/src/catalog.rs +++ b/crates/pyspark-rs/src/catalog.rs @@ -453,8 +453,8 @@ impl PyCatalog { /// Create an external table from a data source: deprecated, as PySpark's, which warns and /// creates the table as `createTable` does. - #[pyo3(signature = (tableName, path=None, source=None, schema=None, **options))] - #[allow(non_snake_case)] + #[pyo3(signature = (tableName, path=None, source=None, schema=None, description=None, **options))] + #[allow(non_snake_case, clippy::too_many_arguments)] fn createExternalTable( &self, py: Python<'_>, @@ -462,6 +462,7 @@ impl PyCatalog { path: Option<&str>, source: Option<&str>, schema: Option<&Bound<'_, PyAny>>, + description: Option<&str>, options: Option<&Bound<'_, pyo3::types::PyDict>>, ) -> PyResult { py.import("warnings")?.call_method1( @@ -472,7 +473,7 @@ impl PyCatalog { py.get_type::(), ), )?; - self.createTable(py, tableName, path, source, schema, None, options) + self.createTable(py, tableName, path, source, schema, description, options) } /// Register a Python function as a UDF, returning the registered UDF. Deprecated diff --git a/crates/pyspark-rs/src/readwriter.rs b/crates/pyspark-rs/src/readwriter.rs index df9515d7..8ffb1c3d 100644 --- a/crates/pyspark-rs/src/readwriter.rs +++ b/crates/pyspark-rs/src/readwriter.rs @@ -33,15 +33,7 @@ impl PyDataFrameReader { /// A copy of the current reader with per-call `**options` applied (skipping /// None, lowercasing bools), for a format-specific read. fn with_opts(&self, options: Option<&Bound<'_, PyDict>>) -> PyResult { - let mut r = self.inner.clone(); - if let Some(opts) = options { - for (k, v) in opts.iter() { - if let Some(val) = crate::coerce_option_value(&v)? { - r = r.option(&k.str()?.to_string(), &val); - } - } - } - Ok(r) + apply_options(self.inner.clone(), options, false) } /// Keep the options of a format-specific read on this reader, as PySpark does @@ -183,6 +175,27 @@ pub(crate) fn bucket_count(num_buckets: &Bound<'_, PyAny>) -> PyResult { } } +/// The reader `r` with the `options` of a call, lowercasing bools. A None value unsets its +/// option when `unset_none`, as PySpark's `DataFrameReader.option`, `options` and `load` do, +/// and is skipped otherwise, as the `_set_opts` of its format methods does. +fn apply_options( + mut r: DataFrameReader, + options: Option<&Bound<'_, PyDict>>, + unset_none: bool, +) -> PyResult { + if let Some(opts) = options { + for (k, v) in opts.iter() { + let key = k.str()?.to_string(); + match crate::coerce_option_value(&v)? { + Some(val) => r = r.option(&key, &val), + None if unset_none => r = r.unset_option(&key), + None => {} + } + } + } + Ok(r) +} + /// Apply a single reader option (skipping None; bool->"true"/"false"). fn set_opt( r: DataFrameReader, @@ -306,35 +319,28 @@ impl PyDataFrameReader { Ok(slf) } - /// Set a single read option. `None` leaves the option unset (not the string - /// "None"); booleans lowercase to "true"/"false" (reference `to_str` semantics). + /// Set a single read option. `None` unsets the option, as PySpark's does (not the + /// string "None"); booleans lowercase to "true"/"false" (reference `to_str` semantics). fn option<'py>( mut slf: PyRefMut<'py, Self>, key: &str, value: &Bound<'_, PyAny>, ) -> PyResult> { - if let Some(v) = crate::coerce_option_value(value)? { - slf.inner = slf.inner.clone().option(key, &v); - } + slf.inner = match crate::coerce_option_value(value)? { + Some(v) => slf.inner.clone().option(key, &v), + None => slf.inner.clone().unset_option(key), + }; Ok(slf) } /// Set multiple read options. Mirrors reference `DataFrameReader.options(**options)` - /// - keyword args; `None` values are skipped and booleans lowercased. + /// - keyword args; a `None` value unsets the option and booleans lowercase. #[pyo3(signature = (**options))] fn options<'py>( mut slf: PyRefMut<'py, Self>, options: Option<&Bound<'_, PyDict>>, ) -> PyResult> { - let mut map = HashMap::new(); - if let Some(options) = options { - for (k, v) in options.iter() { - if let Some(val) = crate::coerce_option_value(&v)? { - map.insert(k.str()?.to_string(), val); - } - } - } - slf.inner = slf.inner.clone().options(map); + slf.inner = apply_options(slf.inner.clone(), options, true)?; Ok(slf) } @@ -380,7 +386,7 @@ impl PyDataFrameReader { schema: Option<&Bound<'_, PyAny>>, options: Option<&Bound<'_, PyDict>>, ) -> PyResult { - let mut r = self.with_opts(options)?; + let mut r = apply_options(self.inner.clone(), options, true)?; if let Some(f) = format { r = r.format(f); } diff --git a/crates/spark-connect/src/readwriter.rs b/crates/spark-connect/src/readwriter.rs index c1d8ea98..5680f360 100644 --- a/crates/spark-connect/src/readwriter.rs +++ b/crates/spark-connect/src/readwriter.rs @@ -60,6 +60,12 @@ impl DataFrameReader { self } + /// Unset a read option, as PySpark's `DataFrameReader.option(key, None)` does. + pub fn unset_option(mut self, key: &str) -> Self { + self.options.remove(key); + self + } + /// Set multiple options. pub fn options(mut self, options: HashMap) -> Self { self.options.extend(options); diff --git a/python/tests/test_dropin_e2e.py b/python/tests/test_dropin_e2e.py index 157284eb..1713a477 100644 --- a/python/tests/test_dropin_e2e.py +++ b/python/tests/test_dropin_e2e.py @@ -1357,6 +1357,42 @@ def test_reader_schema_keeps_field_metadata(session, tmp_path, ddl): assert stream.schema.fields[0].metadata == {"comment": "the key"} +def test_reader_none_unsets_an_option_and_external_description_is_the_comment(session, tmp_path): + # As PySpark's: DataFrameReader.option, options and load unset an option given None, while + # the format methods skip None and keep it. createExternalTable's description is the + # comment of createTable, not a property. + import warnings + + (tmp_path / "t.csv").write_text("a,b\n1,2\n") + path = str(tmp_path / "t.csv") + assert session.read.option("header", "true").csv(path).columns == ["a", "b"] + unset = session.read.option("header", "true").option("header", None) + assert unset.format("csv").load(path).columns == ["_c0", "_c1"] + unset = session.read.option("header", "true").options(header=None) + assert unset.format("csv").load(path).columns == ["_c0", "_c1"] + unset = session.read.option("header", "true").format("csv") + assert unset.load(path, header=None).columns == ["_c0", "_c1"] + kept = session.read.option("header", "true") + assert kept.csv(path, header=None).columns == ["a", "b"] + assert kept.format("csv").load(path).columns == ["a", "b"] + session.range(2).write.mode("overwrite").parquet(str(tmp_path / "ext")) + session.sql("DROP TABLE IF EXISTS i3_described_external") + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", FutureWarning) + session.catalog.createExternalTable( + "i3_described_external", path=str(tmp_path / "ext"), source="parquet", + description="the table") + assert session.catalog.getTable("i3_described_external").description == "the table" + comments = session.sql("DESCRIBE TABLE EXTENDED i3_described_external").filter( + "col_name = 'Comment'").collect() + assert [tuple(r)[:2] for r in comments] == [("Comment", "the table")] + keys = [r.key for r in session.sql("SHOW TBLPROPERTIES i3_described_external").collect()] + assert not any("description" in key for key in keys) + finally: + session.sql("DROP TABLE IF EXISTS i3_described_external") + + def test_reader_writer_and_catalog_return_the_results_of_pyspark(session, tmp_path): # As PySpark's: csv, json and xml of a DataFrame of strings parse them, the readers load # every path, and createTable creates the table on the call, of its schema and options.