dbtsai commented on code in PR #109:
URL:
https://github.com/apache/spark-connect-rust/pull/109#discussion_r4404604936
##########
crates/spark-connect/src/session.rs:
##########
@@ -618,22 +618,17 @@ fn rows_to_arrow_ipc(rows: &[Row], schema: &DataType) ->
Result<Vec<u8>> {
_ => return Err(SparkError::connect_msg("Schema is not a struct
type")),
};
- // Coerce each value to its declared field type so the built arrays match
the
- // Arrow schema (e.g. a Python int decodes as Long but "a int" wants
Int32).
- let coerced: Vec<Row> = rows
- .iter()
- .map(|r| {
- let vals: Vec<Value> = fields
- .iter()
- .enumerate()
- .map(|(i, f)| coerce_value(r.get(i), &f.data_type))
- .collect();
- Row::new(r.fields().to_vec(), vals)
- })
- .collect();
+ // Preserve the width check previously performed by each coerced Row.
Review Comment:
Changed in f26ea7e: the core width check now returns `SparkError`, with a
positional-conversion comment. I also added validation before `Row::new` in the
Python wrapper, where explicit-schema input could otherwise still panic. Rust
short/long-row cases and Python inferred/names-only/StructType regressions pass.
##########
crates/spark-connect/src/session.rs:
##########
@@ -1613,3 +1611,277 @@ mod tests {
assert!(tags.contains(&"test".to_string()));
}
}
+
+#[cfg(test)]
+mod input_ipc_tests {
+ // Keep IPC encoding fixtures separate from the session API tests above.
+ use super::*;
+ use arrow::array::{
+ BinaryArray, BooleanArray, Float32Array, Int16Array, Int32Array,
Int64Array, Int8Array,
+ StringArray, StructArray,
+ };
+ use arrow::ipc::reader::StreamReader;
+ use arrow::record_batch::RecordBatch;
+ use std::io::Cursor;
+
+ fn field(name: &str, data_type: DataType) -> crate::types::StructField {
+ crate::types::StructField {
+ name: name.to_string(),
+ data_type,
+ nullable: true,
+ metadata: Default::default(),
+ }
+ }
+
+ fn text_type() -> DataType {
+ DataType::String {
+ collation: "UTF8_BINARY".to_string(),
+ }
+ }
+
+ fn schema(types: Vec<DataType>) -> DataType {
+ DataType::Struct {
+ fields: types
+ .into_iter()
+ .enumerate()
+ .map(|(i, dt)| field(&format!("schema_{i}"), dt))
+ .collect(),
+ }
+ }
+
+ fn row(values: Vec<Value>) -> Row {
+ // Top-level conversion is positional, even when input and schema
names differ.
+ Row::new(
+ (0..values.len()).map(|i| format!("input_{i}")).collect(),
+ values,
+ )
+ }
+
+ fn read_batch(rows: &[Row], schema: &DataType) -> RecordBatch {
+ let data = rows_to_arrow_ipc(rows, schema).unwrap();
+ let mut reader = StreamReader::try_new(Cursor::new(data),
None).unwrap();
+ let batch = reader.next().unwrap().unwrap();
+ assert!(reader.next().is_none());
+ assert_eq!(batch.num_rows(), rows.len());
+ batch
+ }
+
+ #[test]
+ fn input_ipc_preserves_scalar_coercion_and_nulls() {
+ // Compatibility check for this optimization: integer overflow
currently
Review Comment:
Removed the out-of-range integer cases in f26ea7e, retaining in-range
conversion and null coverage. The duplicate-name bug expectation is also gone;
the replacement checks independent Float/Double conversion from the original
source value.
##########
crates/spark-connect/src/session.rs:
##########
@@ -815,12 +809,16 @@ fn build_arrow_array(
if rows.is_empty() {
return Ok(Arc::new(NullArray::new(0)));
}
- let column: Vec<Value> = rows
- .iter()
- .map(|r| r.get(field_idx).cloned().unwrap_or(Value::Null))
- .collect();
match target {
- Some(dt) => values_to_arrow(&column, dt),
+ Some(dt) => {
+ // Coerce only this column, avoiding a second complete row matrix
and
+ // a deep clone of each coerced value before Arrow construction.
+ let column: Vec<Value> = rows
+ .iter()
+ .map(|r| coerce_value(r.get(field_idx), dt))
Review Comment:
Removed the coercion pass in f26ea7e; values now go directly to the
schema-driven encoder, so nested struct names are no longer copied by coercion.
The historical comment is gone too. All 392 library tests and 11 independent
Python fixture comparisons pass, and integration targets compile with and
without `wasm-udf`. Full live Spark parity is still pending in CI; the fixture
checks do not establish that. The description now distinguishes the old
benchmark results from this revised implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
