[SPARK-61179][CONNECT][PYTHON] Unset a DataFrameReader option of None, and take the description of Catalog.createExternalTable, in the Python client as PySpark does

crates/pyspark-rs/src/catalog.rs

@@ -453,15 +453,16 @@ 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<'_>,
         tableName: &str,
         path: Option<&str>,
         source: Option<&str>,
         schema: Option<&Bound<'_, PyAny>>,
+        description: Option<&str>,
         options: Option<&Bound<'_, pyo3::types::PyDict>>,
     ) -> PyResult<PyDataFrame> {
         py.import("warnings")?.call_method1(
@@ -472,7 +473,7 @@ impl PyCatalog {
                 py.get_type::<pyo3::exceptions::PyFutureWarning>(),
             ),
         )?;
-        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

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<DataFrameReader> {
-        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<i32> {
     }
 }
 
+/// 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<DataFrameReader> {
+    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<PyRefMut<'py, Self>> {
-        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<PyRefMut<'py, Self>> {
-        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<PyDataFrame> {
-        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);
         }

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<String, String>) -> Self {
         self.options.extend(options);

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.