comphead commented on code in PR #7106:
URL: https://github.com/apache/datafusion-comet/pull/7106#discussion_r4332014144
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -876,7 +878,78 @@ fn cast_binary_to_string<O: OffsetSizeTrait>(
let num_rows = input.len();
let offsets = input.value_offsets();
- let value_bytes = offsets[num_rows].as_usize() - offsets[0].as_usize();
+ let start = offsets[0].as_usize();
+ let end = offsets[num_rows].as_usize();
+ // Bytes of the non-null rows. A null row can still span bytes, for
example after `nullif`.
+ let live_bytes = match input.nulls().filter(|nulls| nulls.null_count() >
0) {
+ Some(nulls) => nulls
+ .valid_indices()
+ .map(|i| offsets[i + 1].as_usize() - offsets[i].as_usize())
+ .sum(),
+ None => end - start,
+ };
+
+ // Check the bytes the rows span once. If they are valid UTF-8, reuse the
input's value buffer,
+ // or copy them when they fill less than half of its allocation; otherwise
copy the rows in the
+ // valid prefix and decode the rest. Skip this when null rows span more
than half of those
+ // bytes: decoding the live rows one at a time then reads less and returns
a compact buffer
+ // instead of keeping the null rows' bytes alive.
+ let mut valid_prefix = "";
+ if matches!(
+ spark_cast_options.binary_output_style,
+ None | Some(BinaryOutputStyle::Utf8)
+ ) && 2 * live_bytes >= end - start
+ {
+ let data = input.value_data();
+ match std::str::from_utf8(&data[start..end]) {
+ Ok(valid) => {
+ // An offset splits a code point only if it points at a
continuation byte.
+ if offsets.iter().all(|offset| {
+ let offset = offset.as_usize();
+ offset == end || !matches!(data[offset], 0x80..=0xBF)
+ }) {
+ // A slice of a large array shares its whole allocation,
and memory accounting
+ // such as a sort's charges all of it to every batch that
holds it. Copy the
+ // bytes into a compact buffer then, without decoding them
again.
+ let (offsets, values) = if 2 * (end - start) >=
input.values().capacity() {
+ // Rebase a slice's offsets onto the narrowed value
buffer.
+ let offsets = if start == 0 {
+ input.offsets().clone()
+ } else {
+
OffsetBuffer::from_lengths(input.offsets().lengths())
+ };
+ (
+ offsets,
+ input.values().slice_with_length(start, end -
start),
+ )
+ } else {
+ (
+
OffsetBuffer::from_lengths(input.offsets().lengths()),
+ Buffer::from_slice_ref(&data[start..end]),
+ )
+ };
+ // SAFETY: the value buffer is valid UTF-8 and every
offset falls on a char
+ // boundary, which is what `GenericStringArray::try_new`
checks.
+ let strings = unsafe {
+ GenericStringArray::<O>::new_unchecked(
Review Comment:
The compact-copy branch rebuilds the offsets with
`OffsetBuffer::from_lengths(input.offsets().lengths())` even when `start == 0`,
while the reuse branch above already clones `input.offsets()` for that case.
Only the values buffer differs between the two branches, so the offsets could
be computed once and shared. That avoids an O(rows) allocation and checked-add
pass for unsliced arrays and shortens the nested block. Separately, the doc
comment on `cast_binary_to_string` a few lines up still says that without a
binary output style "the bytes are reinterpreted as a string as-is". As far as
I can tell the function now decodes and can return an array that shares the
input's value buffer, so that sentence seems worth updating while this function
is being reworked.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,28 @@ fn parquet_convert_array_impl(
Ok(Arc::new(string_array))
}
+ // Unannotated BINARY read as a string. Spark's cast replaces
malformed UTF-8 with
+ // U+FFFD, where Arrow's safe cast below would return NULL. Top-level
columns don't
+ // reach this arm (#7032). The reader keeps a file's ARROW:schema
types, so an Arrow
+ // writer's `LargeBinary` or `BinaryView` arrives as is; Spark's cast
takes only
+ // `Binary`, and converting `LargeBinary` reuses its value buffer.
+ (Binary | LargeBinary | BinaryView, Utf8) => {
+ let array = match from_type {
+ Binary => array,
+ _ => cast_with_options(&array, &Binary, &PARQUET_OPTIONS)?,
+ };
+ let mut cast_options =
+ SparkCastOptions::new(parquet_options.eval_mode,
&parquet_options.timezone);
+ cast_options.allow_cast_unsigned_ints =
parquet_options.allow_cast_unsigned_ints;
+ cast_options.is_adapting_schema = true;
+
+ match spark_cast(ColumnarValue::Array(array), to_type,
&cast_options)? {
+ ColumnarValue::Array(array) => Ok(array),
+ ColumnarValue::Scalar(_) => Err(DataFusionError::Internal(
+ "Spark cast returned a scalar for an array
input".to_string(),
+ )),
+ }
+ }
// If Arrow cast supports the cast, delegate the cast to Arrow
Review Comment:
For `(Binary, Utf8)` the call reaches `cast_binary_to_string`, which reads
only `binary_output_style`. As far as I can tell `eval_mode`, `timezone`,
`allow_cast_unsigned_ints` and `is_adapting_schema` have no effect on this
pair, so copying them from `parquet_options` adds nothing, and
`SparkCastOptions::new_without_timezone(EvalMode::Legacy)` would say the same
thing. The `ColumnarValue::Scalar(_) => Err(Internal)` arm also looks
unreachable for an array input. Could the call use `.into_array(len)`, or call
the kernel directly? I did not run this.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala:
##########
@@ -146,6 +146,75 @@ abstract class ParquetReadSuite extends CometTestBase {
}
}
+ test("native scan decodes invalid UTF-8 in nested BINARY read as string") {
Review Comment:
Neither fixture row has a NULL, so there is no null list element, struct
field or map value, and no null list, struct or map row. The Rust cast tests
cover nulls inside the kernel, but nothing here exercises nulls through the
list, struct and map plumbing in `parquet_convert_array_impl` (offsets and
parent nulls). A third row with a null element, a null struct field, a null map
value and a null whole container would pin that. A null-bearing valid row ahead
of the invalid one would also give a batch that mixes nulls, valid values and
an invalid value.
--
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]
