sunchao commented on code in PR #7106:
URL: https://github.com/apache/datafusion-comet/pull/7106#discussion_r4324642892
##########
native/spark-expr/src/conversion_funcs/cast.rs:
##########
@@ -876,7 +878,63 @@ 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;
+ // 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)
+ }) {
+ // 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())
+ };
+ // 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(
+ offsets,
+ input.values().slice_with_length(start, end -
start),
Review Comment:
[P2] [P2] Compact slices that retain oversized backing allocations. For a
grouped binary key cast to string before a partition-local sort, such as
`SELECT CAST(k AS STRING) AS s FROM t GROUP BY k SORT BY s`, aggregate output
can arrive as slices of one large array. `slice_with_length` narrows the view
but retains the original allocation and its capacity. The previous cast copied
each slice into a compact buffer. The new output makes DataFusion’s sorter
charge the large backing allocation for every incoming batch, causing avoidable
spilling. With 262,144 valid 64-byte keys, 8,192-row batches and a 64 MiB pool,
the base produced zero spills while this head produced 16 and spilled every
output row. Each 512 KiB slice retained a 16 MiB value allocation. Could the
fast path compact slices when retained allocation capacity is
disproportionately larger than their visible payload, while retaining zero-copy
conversion for appropriately sized arrays?
Evidence: Ran `LD_LIBRARY_PATH=/usr/lib/jvm/java-21-openjdk-amd64/lib/server
/tmp/pr7106-4a62-this-review-actual-cast-memory`. Source:
`/tmp/pr7106-4a62-this-review-actual-cast-memory.rs`. The bounded DataFusion
55.2.0 probe executes AggregateExec → ProjectionExec → SortExec over 262,144
distinct valid 64-byte binary keys. The projection wraps the exact-head
compiled Comet `spark_cast`; the baseline helper was verified byte-for-byte
against base 39380cca813bc0b861353de2cb53af960dd2a87f. Both returned identical
262,144-row sorted results. Base value capacity per batch was 524,288 bytes
with spill_count=0. Head capacity was 16,777,216 bytes with spill_count=16 and
all rows spilled. Arrow 59.3.0 `Buffer::slice_with_length` clones the
underlying allocation, and DataFusion’s sorter reserves buffer capacity
separately for each batch. A disposable variant gating reuse on retained
capacity reproduced zero spills with identical output. This measures allocation
accounting and actual s
pill behavior, not whole-query performance.
--
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]
