Skip to content

Commit aee5e06

Browse files
authored
fix: zero sliced boolean offsets at every level before exporting to the JVM (apache#6339)
Arrow Java's C Data import ignores ArrowArray.offset at every level (apache/arrow-java#88). arrow-rs folds a slice into the buffers for every Spark type except boolean, which keeps its bit offset, so a sliced boolean reached the JVM read from bit 0. The fix for apache#2051 only checked the top level of each output column, so a boolean inside a sliced struct, list or map came back wrong, and inputs to the JVM UDF bridge were never checked at all. Add zero_offsets to datafusion-comet-common. It returns the array unchanged when every offset is already zero, and otherwise re-slices boolean bitmaps to start at bit 0 at every level, sharing every other buffer. move_to_spark now applies it to every executed batch and decoded shuffle block, replacing the top-level take in prepare_output, and JvmScalarUdfExpr applies it to the bridge inputs. Closes apache#6288.
1 parent f0c8e18 commit aee5e06

8 files changed

Lines changed: 434 additions & 29 deletions

File tree

‎docs/source/contributor-guide/ffi.md‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,6 +256,18 @@ pub extern "system" fn Java_..._exportVector(
256256
}
257257
```
258258

259+
### Array Offsets
260+
261+
Arrow Java's C Data import ignores `ArrowArray.offset` at every level
262+
([apache/arrow-java#88](https://github.com/apache/arrow-java/issues/88)) and reads each buffer from its start. arrow-rs
263+
folds a slice into the buffers for almost every type, but a sliced `BooleanArray` keeps its bit offset, and a struct
264+
exports offset 0 even when its children are sliced. So every array native exports to the JVM first goes through
265+
`zero_offsets` (in `native/common/src/ffi_offsets.rs`), which re-slices boolean bitmaps to start at bit 0 at every
266+
level and shares every other buffer. `move_to_spark` applies it to executed batches and decoded shuffle blocks, and
267+
`JvmScalarUdfExpr` applies it to the inputs of the JVM UDF bridge. A new native to JVM export path has to call it too,
268+
or sliced booleans reach the JVM misaligned
269+
([#6288](https://github.com/apache/datafusion-comet/issues/6288)).
270+
259271
### Wrapper Object Lifecycle (Native → JVM)
260272

261273
```

‎native/common/src/ffi_offsets.rs‎

Lines changed: 255 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,255 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
//! Zeroing array offsets before an array is exported to the JVM.
19+
//!
20+
//! Arrow Java's C Data importer ignores `ArrowArray.offset` at every level
21+
//! (<https://github.com/apache/arrow-java/issues/88>) and reads each buffer from its start. arrow-rs
22+
//! folds a slice into the buffers for almost every type, so a sliced `Int64Array`, `StringArray` or
23+
//! `StructArray` exports offset 0 and nothing is lost. `BooleanArray` is the exception: it keeps its
24+
//! bit offset in `ArrayData::offset`, and the export keeps the validity bitmap at that same offset,
25+
//! so the JVM reads both the values and the nulls of a sliced boolean from bit 0. A struct exports
26+
//! offset 0 even when its children are sliced, so checking only the top level misses a boolean
27+
//! child (<https://github.com/apache/datafusion-comet/issues/6288>).
28+
29+
use arrow::array::{ArrayData, MutableArrayData};
30+
use arrow::datatypes::DataType;
31+
use arrow::error::ArrowError;
32+
use std::borrow::Cow;
33+
34+
/// `data` with a zero offset at every level, children and dictionary values included, so that a
35+
/// consumer that ignores `ArrowArray.offset` reads the right rows. Apply it to every array exported
36+
/// to the JVM.
37+
///
38+
/// Returns `data` itself when every offset is already zero, which is the common case. A sliced
39+
/// boolean gets its values bitmap re-sliced to start at bit 0, which shares the buffer when the
40+
/// offset is a whole number of bytes and copies the bitmap otherwise. Every other buffer is shared.
41+
/// The validity bitmap needs nothing here, because `FFI_ArrowArray::new` realigns it to the
42+
/// exported offset.
43+
pub fn zero_offsets(data: &ArrayData) -> Result<Cow<'_, ArrayData>, ArrowError> {
44+
if !has_nonzero_offset(data) {
45+
return Ok(Cow::Borrowed(data));
46+
}
47+
let zeroed = rewrite(data)?;
48+
debug_assert!(!has_nonzero_offset(&zeroed));
49+
Ok(Cow::Owned(zeroed))
50+
}
51+
52+
/// Whether `data` or anything below it has a non-zero offset. A dictionary's values are its only
53+
/// child in `ArrayData`, so they are covered too.
54+
fn has_nonzero_offset(data: &ArrayData) -> bool {
55+
data.offset() != 0 || data.child_data().iter().any(has_nonzero_offset)
56+
}
57+
58+
fn rewrite(data: &ArrayData) -> Result<ArrayData, ArrowError> {
59+
let data = match (data.offset(), data.data_type()) {
60+
(0, _) => data.clone(),
61+
(offset, DataType::Boolean) => {
62+
let values = data.buffers()[0].bit_slice(offset, data.len());
63+
// SAFETY: the same bits, now starting at bit 0. The length and nulls are unchanged.
64+
unsafe {
65+
data.clone()
66+
.into_builder()
67+
.offset(0)
68+
.buffers(vec![values])
69+
.build_unchecked()
70+
}
71+
}
72+
// arrow-rs keeps an offset only for booleans and run-end encoded arrays, and Spark has no
73+
// run-end encoded type, so this is not expected to run. A copy starts at offset 0.
74+
_ => {
75+
let mut copy = MutableArrayData::new(vec![data], false, data.len());
76+
copy.try_extend(0, 0, data.len())?;
77+
copy.freeze()
78+
}
79+
};
80+
if !data.child_data().iter().any(has_nonzero_offset) {
81+
return Ok(data);
82+
}
83+
let children = data
84+
.child_data()
85+
.iter()
86+
.map(|child| zero_offsets(child).map(Cow::into_owned))
87+
.collect::<Result<Vec<_>, _>>()?;
88+
// SAFETY: each child holds the same values as before, with its offsets folded into its buffers.
89+
Ok(unsafe { data.into_builder().child_data(children).build_unchecked() })
90+
}
91+
92+
#[cfg(test)]
93+
mod tests {
94+
use super::*;
95+
use arrow::array::{
96+
Array, ArrayRef, BooleanArray, DictionaryArray, Int32Array, Int64Array, Int8Array,
97+
ListArray, MapArray, StringArray, StructArray,
98+
};
99+
use arrow::buffer::OffsetBuffer;
100+
use arrow::datatypes::{Field, Fields, Int8Type};
101+
use arrow::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema};
102+
use std::sync::Arc;
103+
104+
/// Nullable booleans with nulls and values that differ from their neighbours, so that reading
105+
/// from the wrong bit shows up in both.
106+
fn booleans(len: usize) -> BooleanArray {
107+
(0..len)
108+
.map(|i| (i % 5 != 0).then_some(i % 3 == 0))
109+
.collect()
110+
}
111+
112+
fn assert_exported_offsets_are_zero(array: &FFI_ArrowArray) {
113+
assert_eq!(array.offset(), 0);
114+
(0..array.num_children()).for_each(|i| assert_exported_offsets_are_zero(array.child(i)));
115+
if let Some(dictionary) = array.dictionary() {
116+
assert_exported_offsets_are_zero(dictionary);
117+
}
118+
}
119+
120+
/// Exports `data` the way Comet hands it to the JVM, checks that the export has offset 0 at
121+
/// every level, and checks that importing it back gives the same values.
122+
fn assert_exports_aligned(data: &ArrayData) {
123+
let zeroed = zero_offsets(data).unwrap();
124+
let schema = FFI_ArrowSchema::try_from(data.data_type()).unwrap();
125+
let exported = FFI_ArrowArray::new(&zeroed);
126+
assert_exported_offsets_are_zero(&exported);
127+
let imported = unsafe { from_ffi(exported, &schema) }.unwrap();
128+
assert_eq!(imported, *data);
129+
}
130+
131+
#[test]
132+
fn unsliced_arrays_are_returned_as_is() {
133+
let columns: Vec<ArrayRef> = vec![
134+
Arc::new(booleans(20)),
135+
// Slicing folds into the buffers for every type but boolean.
136+
Arc::new(Int64Array::from_iter_values(0..20).slice(3, 10)),
137+
Arc::new(StringArray::from_iter_values((0..20).map(|i| i.to_string())).slice(3, 10)),
138+
];
139+
for column in columns {
140+
let data = column.to_data();
141+
assert!(matches!(zero_offsets(&data).unwrap(), Cow::Borrowed(_)));
142+
}
143+
}
144+
145+
#[test]
146+
fn sliced_booleans_start_at_bit_zero() {
147+
let array = booleans(100);
148+
// A whole number of bytes in, and part of one.
149+
for offset in [8, 17] {
150+
let data = array.slice(offset, 40).to_data();
151+
assert_exports_aligned(&data);
152+
153+
let zeroed = zero_offsets(&data).unwrap();
154+
assert_eq!(zeroed.offset(), 0);
155+
if offset % 8 == 0 {
156+
// Re-slicing on a byte boundary shares the bitmap.
157+
assert_eq!(
158+
zeroed.buffers()[0].as_ptr(),
159+
data.buffers()[0].as_ptr().wrapping_add(offset / 8)
160+
);
161+
}
162+
}
163+
}
164+
165+
#[test]
166+
fn sliced_boolean_struct_children_start_at_bit_zero() {
167+
let fields = Fields::from(vec![
168+
Field::new("b", DataType::Boolean, true),
169+
Field::new("k", DataType::Int64, false),
170+
]);
171+
let array = StructArray::new(
172+
fields,
173+
vec![
174+
Arc::new(booleans(100)),
175+
Arc::new(Int64Array::from_iter_values(0..100)),
176+
],
177+
None,
178+
);
179+
let data = array.slice(17, 40).to_data();
180+
assert_eq!(data.offset(), 0);
181+
assert_eq!(data.child_data()[0].offset(), 17);
182+
assert_exports_aligned(&data);
183+
184+
// The Int64 child is passed through, not copied.
185+
let zeroed = zero_offsets(&data).unwrap();
186+
assert_eq!(
187+
zeroed.child_data()[1].buffers()[0].as_ptr(),
188+
data.child_data()[1].buffers()[0].as_ptr()
189+
);
190+
}
191+
192+
#[test]
193+
fn sliced_booleans_in_a_list_of_structs_start_at_bit_zero() {
194+
let fields = Fields::from(vec![Field::new("b", DataType::Boolean, true)]);
195+
let structs = StructArray::new(fields.clone(), vec![Arc::new(booleans(100))], None);
196+
// A list slice moves only the offsets, so the list has to be built over a sliced struct,
197+
// which is what collect_list returns for a group with a single run of rows.
198+
let list = ListArray::new(
199+
Arc::new(Field::new_struct("item", fields, true)),
200+
OffsetBuffer::from_lengths([10, 0, 20]),
201+
Arc::new(structs.slice(5, 30)),
202+
None,
203+
);
204+
assert_exports_aligned(&list.to_data());
205+
}
206+
207+
#[test]
208+
fn sliced_boolean_map_values_start_at_bit_zero() {
209+
let entries = StructArray::new(
210+
Fields::from(vec![
211+
Field::new("key", DataType::Utf8, false),
212+
Field::new("value", DataType::Boolean, true),
213+
]),
214+
vec![
215+
Arc::new(StringArray::from_iter_values(
216+
(0..100).map(|i| i.to_string()),
217+
)),
218+
Arc::new(booleans(100)),
219+
],
220+
None,
221+
);
222+
let map = MapArray::try_new(
223+
Arc::new(Field::new_struct(
224+
"entries",
225+
entries.fields().clone(),
226+
false,
227+
)),
228+
OffsetBuffer::from_lengths([3, 7, 0, 10]),
229+
entries.slice(33, 20),
230+
None,
231+
false,
232+
)
233+
.unwrap();
234+
assert_exports_aligned(&map.to_data());
235+
}
236+
237+
#[test]
238+
fn sliced_boolean_dictionary_values_start_at_bit_zero() {
239+
let dictionary = DictionaryArray::<Int8Type>::new(
240+
Int8Array::from(vec![Some(2), None, Some(0), Some(1), Some(2)]),
241+
Arc::new(booleans(20).slice(3, 3)),
242+
);
243+
assert_exports_aligned(&dictionary.to_data());
244+
}
245+
246+
#[test]
247+
fn other_types_with_an_offset_are_copied() {
248+
// arrow-rs never builds one of these, but an ArrayData can carry an offset for any type.
249+
let data = Int32Array::from_iter((0..20).map(|i| (i % 4 != 0).then_some(i)))
250+
.to_data()
251+
.slice(5, 10);
252+
assert_eq!(data.offset(), 5);
253+
assert_exports_aligned(&data);
254+
}
255+
}

‎native/common/src/lib.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
// under the License.
1717

1818
mod error;
19+
mod ffi_offsets;
1920
mod query_context;
2021
mod schema;
2122
pub mod struct_nulls;
@@ -24,6 +25,7 @@ mod utf8;
2425
mod utils;
2526

2627
pub use error::{decimal_overflow_error, SparkError, SparkErrorWithContext, SparkResult};
28+
pub use ffi_offsets::zero_offsets;
2729
pub use query_context::{create_query_context_map, QueryContext, QueryContextMap};
2830
pub use schema::{cast_and_stamp_schema, widen_nested_nullability};
2931
pub use struct_nulls::{child_with_parent_nulls, children_with_parent_nulls};

‎native/core/src/execution/jni_api.rs‎

Lines changed: 4 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -28,8 +28,7 @@ use crate::{
2828
};
2929
use std::collections::HashSet;
3030

31-
use arrow::array::{Array, RecordBatch, UInt32Array};
32-
use arrow::compute::{take, TakeOptions};
31+
use arrow::array::{Array, RecordBatch};
3332
use arrow::datatypes::DataType as ArrowDataType;
3433
use datafusion::common::{DataFusionError, Result as DataFusionResult};
3534
use datafusion::execution::disk_manager::DiskManagerMode;
@@ -984,27 +983,9 @@ fn prepare_output(
984983
let array_ref = results.get(i).ok_or(CometError::IndexOutOfBounds(i))?;
985984
let field = output_schema.field(i);
986985

987-
if array_ref.offset() != 0 {
988-
// https://github.com/apache/datafusion-comet/issues/2051
989-
// Bug with non-zero offset FFI, so take to a new array which will have an offset of 0.
990-
// We expect this to be a cold code path, hence the check_bounds: true and assert_eq.
991-
let indices = UInt32Array::from((0..num_rows as u32).collect::<Vec<u32>>());
992-
let new_array = take(
993-
array_ref,
994-
&indices,
995-
Some(TakeOptions { check_bounds: true }),
996-
)?;
997-
998-
assert_eq!(new_array.offset(), 0);
999-
1000-
new_array
1001-
.to_data()
1002-
.move_to_spark(field, array_addrs[i], schema_addrs[i])?;
1003-
} else {
1004-
array_ref
1005-
.to_data()
1006-
.move_to_spark(field, array_addrs[i], schema_addrs[i])?;
1007-
}
986+
array_ref
987+
.to_data()
988+
.move_to_spark(field, array_addrs[i], schema_addrs[i])?;
1008989
i += 1;
1009990
}
1010991
}

0 commit comments

Comments
 (0)