[SPARK-60720][CONNECT] Encode guarded pandas string columns directly in Rust

crates/pyspark-rs/src/lib.rs

@@ -36,6 +36,7 @@ mod group;
 mod infer;
 mod ml;
 mod observation;
+mod pandas_input;
 mod pipelines;
 mod profiler;
 mod pyarrow_export;
@@ -130,6 +131,8 @@ fn _pyspark(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(types::parse_datatype_json, m)?)?;
     m.add_function(wrap_pyfunction!(column::column_reference_name, m)?)?;
 
+    m.add_function(wrap_pyfunction!(pandas_input::_pandas_string_ipc, m)?)?;
+
     // Spark Declarative Pipelines (SDP) command execution
     m.add_class::<pipelines::PyPipelineRunStream>()?;
     m.add_function(wrap_pyfunction!(

crates/pyspark-rs/src/pandas_input.rs

@@ -0,0 +1,113 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Fused validation and inline encoding of guarded pandas string columns.
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, StringBuilder};
+use arrow::datatypes::{DataType as ArrowType, Field, Schema};
+use arrow::ipc::writer::StreamWriter;
+use arrow::record_batch::RecordBatch;
+use pyo3::prelude::*;
+use pyo3::types::{PyBytes, PyString};
+
+use crate::types::PyStructType;
+
+/// The Python eligibility helper authenticates pandas storage and native schema.
+/// Validate each object before accepting it: extraction must never call a custom
+/// numeric/string conversion hook or turn a pandas missing sentinel into null.
+#[pyfunction]
+pub(crate) fn _pandas_string_ipc<'py>(
+    py: Python<'py>,
+    columns: Vec<Bound<'py, PyAny>>,
+    schema: PyRef<'py, PyStructType>,
+) -> PyResult<Option<Bound<'py, PyBytes>>> {
+    if columns.len() != schema.fields.len() {
+        return Ok(None);
+    }
+    // Reject unsupported objects before copying any string data. Otherwise a
+    // late subclass would pay for most of an Arrow build before the row path.
+    // Keep this pass to exact type/null checks; UTF-8 extraction stays below.
+    for (column, field) in columns.iter().zip(&schema.fields) {
+        if !matches!(&field.data_type, spark_connect::types::DataType::String { collation }
+            if collation == "UTF8_BINARY")
+        {
+            return Ok(None);
+        }
+        for value in column.try_iter()? {
+            let value = value?;
+            if !(value.is_none() && field.nullable || value.is_exact_instance_of::<PyString>()) {
+                return Ok(None);
+            }
+        }
+    }
+    // A lone surrogate must fall back before allocating partial Arrow buffers.
+    // Drop each temporary UTF-8 bytes object immediately rather than retaining
+    // an additional full copy of every column. Keep the cheap type pass above
+    // so a late subclass does not pay for this UTF-8 traversal.
+    for (column, field) in columns.iter().zip(&schema.fields) {
+        for value in column.try_iter()? {
+            let value = value?;
+            if value.is_none() && field.nullable {
+                continue;
+            }
+            let Ok(value) = value.cast_exact::<PyString>() else {
+                return Ok(None);
+            };
+            if value.encode_utf8().is_err() {
+                return Ok(None);
+            }
+        }
+    }
+    let mut arrays: Vec<ArrayRef> = Vec::with_capacity(columns.len());
+    let mut fields = Vec::with_capacity(columns.len());
+    for (column, field) in columns.iter().zip(&schema.fields) {
+        let mut builder = StringBuilder::with_capacity(column.len()?, column.len()? * 8);
+        for value in column.try_iter()? {
+            let value = value?;
+            if value.is_none() && field.nullable {
+                builder.append_null();
+            } else if let Ok(value) = value.cast_exact::<PyString>() {
+                let Ok(bytes) = value.encode_utf8() else {
+                    return Ok(None);
+                };
+                let Ok(text) = std::str::from_utf8(bytes.as_bytes()) else {
+                    return Ok(None);
+                };
+                // append_value copies while the bytes object and GIL are live.
+                // This avoids to_cow's owned String on the abi3-py39 build.
+                builder.append_value(text);
+            } else {
+                return Ok(None);
+            }
+        }
+        arrays.push(Arc::new(builder.finish()));
+        // Like the row encoder, Spark metadata lives in the logical schema;
+        // Arrow fields carry the name, standard string type and nullability.
+        fields.push(Field::new(&field.name, ArrowType::Utf8, field.nullable));
+    }
+    let bytes = py
+        .detach(move || -> Result<Vec<u8>, arrow::error::ArrowError> {
+            let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), arrays)?;
+            let mut writer = StreamWriter::try_new(Vec::new(), &batch.schema())?;
+            writer.write(&batch)?;
+            writer.into_inner()
+        })
+        .map_err(|error| pyo3::exceptions::PyRuntimeError::new_err(error.to_string()))?;
+    Ok(Some(PyBytes::new(py, &bytes)))
+}

crates/pyspark-rs/src/session.rs

@@ -765,6 +765,54 @@ impl PySparkSession {
         }
         let spec = schema_spec(py, schema)?;
         if is_loaded_instance(py, data, "pandas", "DataFrame")? {
+            // Guarded integer columns share the existing Arrow importer. Schema
+            // validation above and unsupported pandas conversions keep master's
+            // normal ordering and row fallback.
+            let table = py
+                .import("pyspark.sql._pandas_input")?
+                .getattr("prepare_int64_table")?
+                .call1((data, schema))?;
+            if !table.is_none() {
+                let frame =
+                    crate::pyarrow_import::create_dataframe(py, &self.session, &table, schema)?;
+                // pandas applies explicit field names with toDF even though the
+                // local relation already carries them. Empty input stays local.
+                return Ok(match &spec {
+                    Spec::Struct(fields) if !fields.is_empty() && table.len()? > 0 => {
+                        PyDataFrame::new(
+                            frame
+                                .dataframe
+                                .to_df(fields.iter().map(|field| field.name.as_str()).collect()),
+                        )
+                    }
+                    _ => frame,
+                });
+            }
+            let ipc = py
+                .import("pyspark.sql._pandas_input")?
+                .getattr("prepare_string_input")?
+                .call1((data, schema))?;
+            if !ipc.is_none() {
+                let fields = match &spec {
+                    Spec::Struct(fields) => fields,
+                    _ => {
+                        return Err(pyo3::exceptions::PyValueError::new_err(
+                            "string input requires a struct schema",
+                        ));
+                    }
+                };
+                let schema = DataType::Struct {
+                    fields: fields.clone(),
+                };
+                let frame = self
+                    .session
+                    .create_dataframe_from_inline_arrow_ipc(ipc.extract()?, schema);
+                // Match pandas row input: explicit nonempty field names travel
+                // through toDF as well as the local relation's logical schema.
+                return Ok(PyDataFrame::new(
+                    frame.to_df(fields.iter().map(|field| field.name.as_str()).collect()),
+                ));
+            }
             return self.create_from_pandas(py, data, spec);
         }
         self.create_from_local(py, data, spec)

crates/spark-connect/src/session.rs

@@ -360,6 +360,23 @@ impl SparkSession {
         Ok(DataFrame::new(self.clone(), plan))
     }
 
+    /// Internal binding bridge for Arrow IPC already validated and encoded from
+    /// local rows or columns. Retains the row path's inline transport policy.
+    #[doc(hidden)]
+    pub fn create_dataframe_from_inline_arrow_ipc(
+        &self,
+        data: Vec<u8>,
+        schema: DataType,
+    ) -> DataFrame {
+        DataFrame::new(
+            self.clone(),
+            LogicalPlan::LocalRelation {
+                schema: Some(schema),
+                data: Some(data),
+            },
+        )
+    }
+
     /// Create a local DataFrame from Arrow batches without materializing rows.
     ///
     /// `schema` optionally supplies the Spark schema applied by the server. With

docs/architecture.md

@@ -142,6 +142,16 @@ timestamp conversion helpers normalize the table, then arrow-rs imports its
 those batches to IPC for the local relation sent to Spark. No Python rows are
 materialized; casts, buffer alignment, and IPC serialization may still copy.
 Empty tables require an explicit schema, matching PySpark Connect.
+Pandas inputs with only NumPy `int64` or pandas nullable `Int64` columns also
+use this importer, preserving integer precision and null masks without Python
+rows. Small frames (fewer than 1,000 rows) with only NumPy columns retain the
+row path to avoid configuration RPC overhead. Nullable `Int64` columns always
+use Arrow so schema inference does not depend on the first value being non-null.
+This initial path accepts inferred schemas, complete string column names,
+and explicit `StructType` schemas whose fields are all `LongType`. Other dtypes,
+DDL/atomic schemas, schema conversions, NUL names, and duplicate/non-string input names
+retain the existing row conversion path. `verifySchema` remains ignored for
+pandas Arrow input, matching Spark Connect.
 Inputs at or above `spark.sql.session.localRelationCacheThreshold` are serialized
 into independent IPC streams and uploaded as session-scoped cache artifacts.
 The plan carries `ChunkedCachedLocalRelation` hashes instead of embedding the