[SPARK-60677][CONNECT] Cache and propagate DataFrame schemas

crates/pyspark-rs/src/dataframe.rs

@@ -1302,8 +1302,8 @@ impl PyDataFrame {
 
     /// Column names (a property, mirroring `DataFrame.columns`).
     #[getter]
-    fn columns(&self) -> PyResult<Vec<String>> {
-        self.dataframe.columns().to_pyerr()
+    fn columns(&self, py: Python<'_>) -> PyResult<Vec<String>> {
+        py.detach(|| self.dataframe.columns().to_pyerr())
     }
 
     /// Get the last `num` rows as a list of Rows (eager, like pyspark `tail`).
@@ -1374,20 +1374,16 @@ 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 {
-            spark_connect::types::DataType::Struct { fields } => {
-                let dtypes = fields
-                    .iter()
-                    .map(|f| (f.name.clone(), f.data_type.simple_string()))
-                    .collect();
-                Ok(dtypes)
-            }
-            _ => Err(PyErr::new::<pyo3::exceptions::PyTypeError, _>(
-                "Schema is not a struct type",
-            )),
-        }
+    fn dtypes(&self, py: Python<'_>) -> PyResult<Vec<(String, String)>> {
+        let spark_connect::types::DataType::Struct { fields } =
+            py.detach(|| self.dataframe.schema().to_pyerr())?
+        else {
+            unreachable!("schema() validates struct responses")
+        };
+        Ok(fields
+            .into_iter()
+            .map(|field| (field.name, field.data_type.simple_string()))
+            .collect())
     }
 
     /// Additional set operation: unionAll (alias for union).