sunchao commented on code in PR #7106:
URL: https://github.com/apache/datafusion-comet/pull/7106#discussion_r4367567431
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,33 @@ 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)?,
+ };
+ // Binary to Utf8 reads only the binary output style, which a scan
leaves unset.
+ let cast_options =
SparkCastOptions::new_without_timezone(EvalMode::Legacy);
+ let len = array.len();
+ spark_cast(ColumnarValue::Array(array), to_type,
&cast_options)?.into_array(len)
+ }
+ // The same schema can dictionary-encode such a value. Decode each
dictionary value once
+ // through the arm above, then expand the keys, as Arrow's cast would.
+ (Dictionary(_, value_type), Utf8) if matches!(value_type.as_ref(),
Binary | LargeBinary) => {
+ let dictionary = array.as_any_dictionary();
+ let values = parquet_convert_array_impl(
Review Comment:
[P2] Avoid decoding the entire row-group dictionary for every output batch.
A selected batch can reference only a short valid string and still repeatedly
pay to repair a wide malformed entry used elsewhere in the row group. Reuse the
conversion while the same dictionary backs the scan, or decode only referenced
entries.
Evidence: One row group contained `normal` and an 86,016-byte malformed
entry referenced only in its first batch. Selecting the other 31 batches
decoded the unused entry 31 times. Median Parquet read-and-adapt CPU was 1.97
ms on base, 7.90 ms on head and 1.85 ms with dictionary reuse. The Spark query
`SELECT sum(octet_length(s.k)) FROM t WHERE part=1` returned `1523712` on both
revisions. Median native scan time rose from 3.67 to 11.01 ms.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,33 @@ 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)?,
+ };
+ // Binary to Utf8 reads only the binary output style, which a scan
leaves unset.
+ let cast_options =
SparkCastOptions::new_without_timezone(EvalMode::Legacy);
+ let len = array.len();
+ spark_cast(ColumnarValue::Array(array), to_type,
&cast_options)?.into_array(len)
+ }
+ // The same schema can dictionary-encode such a value. Decode each
dictionary value once
+ // through the arm above, then expand the keys, as Arrow's cast would.
+ (Dictionary(_, value_type), Utf8) if matches!(value_type.as_ref(),
Binary | LargeBinary) => {
+ let dictionary = array.as_any_dictionary();
+ let values = parquet_convert_array_impl(
+ Arc::clone(dictionary.values()),
+ to_type,
+ parquet_options,
+ None,
+ )?;
+ Ok(take(values.as_ref(), dictionary.keys(), None)?)
Review Comment:
[P2] `take` flattens the decoded dictionary into one `Utf8` array with
signed 32-bit offsets. Repeated references can fit in the input batch and
exceed that limit after replacement. This aborts even `SUM(size(arr))`, which
only needs list lengths. Split the enclosing batch at parent-row boundaries
using decoded widths, or retain the dictionary until it can be materialized in
bounded batches.
Evidence: A 264,786-byte Parquet file held 8,192 singleton lists referencing
one 88,000-byte dictionary value of `FF` bytes. Each decoded value was 264,000
bytes. At the default batch size of 8,192, Spark 4.1.3 and the base returned
`8192` for `SUM(size(arr))`. The head’s native scan raised `Offset overflow
error: 2147640000`. An Arrow probe independently confirmed the failure during
offset calculation.
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -876,7 +881,80 @@ 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)
+ }) {
+ // Reuse a buffer only when the rows fill at least half of
its allocation. A
+ // slice of a large array shares the whole allocation, and
memory accounting
+ // such as a sort's charges all of it to every batch that
holds it, so the
+ // offsets, values and validity bitmap are copied then,
without decoding the
+ // values again.
+ let fills = |buffer: &Buffer, used: usize| 2 * used >=
buffer.capacity();
Review Comment:
[P2] Account for the owner of an imported buffer before sharing it. Arrow C
Stream gives each child buffer a reference to the root record batch. Retaining
an eight-byte cast key can therefore pin a discarded wide sibling for as long
as a downstream sort holds the projection. Detach values, offsets and validity
for batch-owned inputs, or restrict sharing to independently owned buffers.
Evidence: Ran `SELECT CAST(k AS STRING) AS s, length(blob) AS n FROM wide
SORT BY s` on one Spark 4.1.3 RDD partition with
`spark.comet.convert.rdd.enabled=true`, 128-row batches, 8,192 eight-byte keys
and 4 KiB blobs. Both revisions returned every sorted row correctly. Peak
`CometArrowAllocator` usage rose from 852,544 bytes on base to 37,772,352 on
head. Both released all allocations after collection.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,33 @@ 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)?,
+ };
+ // Binary to Utf8 reads only the binary output style, which a scan
leaves unset.
+ let cast_options =
SparkCastOptions::new_without_timezone(EvalMode::Legacy);
+ let len = array.len();
+ spark_cast(ColumnarValue::Array(array), to_type,
&cast_options)?.into_array(len)
Review Comment:
[P2] Preserve map-key identity when decoding a string inside a key. If two
struct keys contain `FF` and `EF BF BD`, this cast turns both into `{s: �}`.
Looking up the second key then returns the first entry’s value. Preserve the
bytes through key evaluation or route affected map reads through a
byte-preserving Spark path. Falling back only the lookup happens too late.
Evidence: A one-row `map<struct<s:string>,int>` fixture stored `FF → 1`
followed by `EF BF BD → 2`. `SELECT m[named_struct('s','�')]` returned `2` on
Spark 4.1.3 and the base, and `1` on the head, with a JVM `Project` over
`CometNativeScan`. `size(map_from_entries(map_entries(m)))` returned `2` on
Spark and base and raised `DUPLICATED_MAP_KEY` on head. Both inferred and
explicit string schemas exhibited the collision.
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -904,14 +982,17 @@ fn cast_binary_to_string<O: OffsetSizeTrait>(
// the row. Writing to the builder is infallible.
let written = match spark_cast_options.binary_output_style {
// Default CAST(binary AS string) and the UTF8 ToPrettyString
style (Spark 4.0+) both
- // render via `new String(bytes, UTF_8)`. Route through the shared
JVM-compatible lossy
- // decoder so ill-formed bytes become U+FFFD (matching Spark)
instead of being
- // reinterpreted unchecked (UB) or panicking on non-UTF-8 input
(#4488, #4763). The
- // valid-UTF-8 path borrows, so it copies once (into the Arrow
value buffer). Divergence
- // for byte-level round-trips such as CAST(CAST(x AS string) AS
binary) and value
- // identity is documented in the compatibility guide and tracked
by #4764.
+ // render via `new String(bytes, UTF_8)`. Values in the valid
prefix are copied as is;
+ // the rest go through the shared JVM-compatible lossy decoder so
ill-formed bytes
+ // become U+FFFD (matching Spark) instead of being reinterpreted
unchecked (UB) or
+ // panicking on non-UTF-8 input (#4488, #4763). Divergence for
byte-level round-trips
+ // such as CAST(CAST(x AS string) AS binary) and value identity is
documented in the
+ // compatibility guide and tracked by #4764.
None | Some(BinaryOutputStyle::Utf8) => {
- builder.write_str(&decode_utf8_spark_lossy(value))
+ match valid_prefix.get(range[0].as_usize() -
start..range[1].as_usize() - start) {
+ Some(valid) => builder.write_str(valid),
+ None => builder.write_str(&decode_utf8_spark_lossy(value)),
Review Comment:
[P2] Reuse the validated prefix inside the first malformed row. When a wide
multibyte value ends in a truncated character, `valid_prefix.get` rejects the
whole row range and `decode_utf8_spark_lossy(value)` starts again at byte zero.
If the row starts on a verified character boundary, append its known-good
prefix and decode the remaining suffix.
Evidence: Across 32 batches containing one roughly 256 KiB truncated
multibyte value and 8,191 short values each, `SELECT octet_length(CAST(k AS
STRING))` returned identical results. Median `CometProject` compute rose from
26.33 ms on base to 29.88 ms on head. With the wide row first, the isolated
cast measured 612 µs on base, 798 µs on head and 335 µs with partial-row prefix
reuse, with identical validated output.
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -876,7 +881,80 @@ 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()
Review Comment:
[P2] Count live bytes by contiguous validity runs for sparsely nullable
input. One null per 128 short Parquet values makes `valid_indices()` visit
nearly every row to establish that the nulls hide zero bytes. Summing offset
differences at run boundaries preserves the hidden-payload check and avoids
that extra pass on ordinary valid strings.
Evidence: For 262,144 nested eight-byte values in 8,192-row batches,
repeated release probes measured about 8 µs per conversion on base and 21 µs on
head. Changing only the count to `valid_slices()` measured about 8.5 µs with
equal, validated output. Plain Parquet read-and-adapt CPU rose from 2.75–2.80
to 3.11–3.23 ms. The separate Spark scan medians were 5.82 ms on base and 4.97
ms on head, so the demonstrated regression is in conversion and reader CPU.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -330,6 +330,33 @@ 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)?,
+ };
+ // Binary to Utf8 reads only the binary output style, which a scan
leaves unset.
+ let cast_options =
SparkCastOptions::new_without_timezone(EvalMode::Legacy);
+ let len = array.len();
+ spark_cast(ColumnarValue::Array(array), to_type,
&cast_options)?.into_array(len)
Review Comment:
[P2] Compact the newly decoded value buffer when a small replacement pushes
the builder past its initial capacity. One malformed byte in a 512 KiB batch
adds two output bytes, yet the new scan path hands a 1 MiB allocation to the
sorter. Sizing for decoded bytes or shrinking the uniquely owned fallback
buffer before returning it avoids the extra spill pressure.
Evidence: Sorted 262,144 nested 64-byte keys with one malformed final byte
per 8,192-row batch and a 52 MiB pool. Each head batch used 524,290 bytes in a
1,048,576-byte allocation. Head spilled twice and wrote 2,390,680 bytes. Base
returned 32 nulls and had zero spills. A head control with the same
replacements already encoded produced identical strings and zero spills.
Compacting the fallback buffer preserved all values at 524,290 bytes.
--
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]
