Skip to content

Commit 0fb7157

Browse files
committed
style: run cargo fmt
1 parent 905f0b3 commit 0fb7157

4 files changed

Lines changed: 199 additions & 12 deletions

File tree

‎build.log‎

392 Bytes
Binary file not shown.

‎datafusion/common/src/config.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4056,6 +4056,7 @@ impl Display for OutputFormat {
40564056
mod tests {
40574057
#[cfg(feature = "parquet")]
40584058
use crate::assert_contains;
4059+
#[cfg(any(feature = "parquet", feature = "parquet_encryption"))]
40594060
use crate::config::TableParquetOptions;
40604061
use crate::config::{
40614062
ConfigEntry, ConfigExtension, ConfigField, ConfigFileType, ExtensionOptions,

‎datafusion/common/src/rounding.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -254,7 +254,7 @@ where
254254
}
255255
}
256256
_ => {}
257-
};
257+
}
258258
Ok(result)
259259
}
260260

‎datafusion/physical-plan/src/joins/hash_join/exec.rs‎

Lines changed: 197 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -3012,22 +3012,23 @@ fn deduplicate_view_array_buffers<T: ByteViewType>(
30123012
return array.clone();
30133013
}
30143014

3015-
// Use the raw buffer address as the deduplication key. Casting to usize is the
3015+
// Use the raw buffer address and length as the deduplication key. Casting to usize is the
30163016
// idiomatic way to use pointer values as HashMap keys on stable Rust.
30173017
let mut unique_buffers: Vec<arrow::buffer::Buffer> =
30183018
Vec::with_capacity(data_buffers.len());
3019-
let mut pointer_map: HashMap<usize, u32> = HashMap::with_capacity(data_buffers.len());
3019+
let mut pointer_map: HashMap<(usize, usize), u32> =
3020+
HashMap::with_capacity(data_buffers.len());
30203021
let mut index_remap: Vec<u32> = Vec::with_capacity(data_buffers.len());
30213022
let mut has_duplicates = false;
30223023

30233024
for buf in data_buffers.iter() {
3024-
let addr = buf.as_ptr() as usize;
3025-
if let Some(&new_idx) = pointer_map.get(&addr) {
3025+
let key = (buf.as_ptr() as usize, buf.len());
3026+
if let Some(&new_idx) = pointer_map.get(&key) {
30263027
index_remap.push(new_idx);
30273028
has_duplicates = true;
30283029
} else {
30293030
let new_idx = unique_buffers.len() as u32;
3030-
pointer_map.insert(addr, new_idx);
3031+
pointer_map.insert(key, new_idx);
30313032
unique_buffers.push(buf.clone());
30323033
index_remap.push(new_idx);
30333034
}
@@ -7739,12 +7740,18 @@ mod tests {
77397740
let schema =
77407741
Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8View, true)]));
77417742

7742-
let batch1 =
7743-
RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(base_array.clone())])?;
7744-
let batch2 =
7745-
RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(base_array.clone())])?;
7746-
let batch3 =
7747-
RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(base_array.clone())])?;
7743+
let batch1 = RecordBatch::try_new(
7744+
Arc::clone(&schema),
7745+
vec![Arc::new(base_array.clone())],
7746+
)?;
7747+
let batch2 = RecordBatch::try_new(
7748+
Arc::clone(&schema),
7749+
vec![Arc::new(base_array.clone())],
7750+
)?;
7751+
let batch3 = RecordBatch::try_new(
7752+
Arc::clone(&schema),
7753+
vec![Arc::new(base_array.clone())],
7754+
)?;
77487755

77497756
// Before deduplication, concat_batches puts 3 duplicate buffer references in data_buffers
77507757
let concatenated_raw =
@@ -7781,6 +7788,185 @@ mod tests {
77817788
Ok(())
77827789
}
77837790

7791+
#[test]
7792+
fn concat_build_batches_deduplicates_binary_view_buffers() -> Result<()> {
7793+
use arrow::array::BinaryViewBuilder;
7794+
7795+
let mut builder = BinaryViewBuilder::new();
7796+
builder.append_value(b"this is a long binary that exceeds inline size 12");
7797+
builder.append_value(b"another long binary that exceeds inline size 12");
7798+
let base_array: BinaryViewArray = builder.finish();
7799+
let schema = Arc::new(Schema::new(vec![Field::new(
7800+
"b",
7801+
DataType::BinaryView,
7802+
true,
7803+
)]));
7804+
7805+
let batch1 = RecordBatch::try_new(
7806+
Arc::clone(&schema),
7807+
vec![Arc::new(base_array.clone())],
7808+
)?;
7809+
let batch2 = RecordBatch::try_new(
7810+
Arc::clone(&schema),
7811+
vec![Arc::new(base_array.clone())],
7812+
)?;
7813+
let batch3 = RecordBatch::try_new(
7814+
Arc::clone(&schema),
7815+
vec![Arc::new(base_array.clone())],
7816+
)?;
7817+
7818+
let metrics = BuildProbeJoinMetrics::new(0, &ExecutionPlanMetricsSet::new());
7819+
let pool: Arc<dyn MemoryPool> = Arc::new(UnboundedMemoryPool::default());
7820+
let batches = vec![batch1, batch2, batch3];
7821+
let (mut reservation, inputs_reserved) = reserve_inputs(&batches, &pool)?;
7822+
7823+
let batch = concat_build_batches(
7824+
&schema,
7825+
batches,
7826+
false,
7827+
inputs_reserved,
7828+
&mut reservation,
7829+
&metrics,
7830+
)?;
7831+
7832+
let view_arr = batch
7833+
.column(0)
7834+
.as_any()
7835+
.downcast_ref::<BinaryViewArray>()
7836+
.unwrap();
7837+
assert_eq!(view_arr.data_buffers().len(), 1);
7838+
assert_eq!(batch.num_rows(), 6);
7839+
Ok(())
7840+
}
7841+
7842+
/// Regression test: two arrays backed by Buffer objects that share the same
7843+
/// base pointer but have *different* declared lengths (produced via
7844+
/// `Buffer::slice_with_length`) must NOT be collapsed into one buffer entry,
7845+
/// because the longer one covers bytes beyond the shorter one's range.
7846+
/// Values that live in the longer buffer must survive concatenation intact.
7847+
#[test]
7848+
fn concat_build_batches_deduplicates_slices_regression() -> Result<()> {
7849+
use arrow::array::StringViewBuilder;
7850+
use arrow::buffer::Buffer;
7851+
7852+
// Build a two-element array so the builder allocates a single contiguous
7853+
// data buffer large enough to hold both non-inline strings.
7854+
let mut builder = StringViewBuilder::new();
7855+
builder.append_value("this is a long string that exceeds inline size 12");
7856+
builder.append_value("another long string that exceeds inline size 12");
7857+
let base_array: StringViewArray = builder.finish();
7858+
7859+
// Pull out the single data buffer Arrow produced.
7860+
let data_buffers = base_array.data_buffers();
7861+
assert_eq!(data_buffers.len(), 1, "expected one backing data buffer");
7862+
let full_buf: &Buffer = &data_buffers[0];
7863+
let full_len = full_buf.len();
7864+
7865+
// Slice the same allocation to two different declared lengths. Both
7866+
// handles start at offset 0, so `buf.as_ptr()` is identical, but
7867+
// `buf.len()` differs. The deduplication key is (ptr, len), so these
7868+
// are distinct keys — the longer buffer must not be discarded.
7869+
//
7870+
// Derive the split point from the first view's actual byte range so
7871+
// the short buffer provably covers only the first string.
7872+
let view0 = ByteView::from(base_array.views()[0]);
7873+
let first_string_end = view0.offset as usize + view0.length as usize;
7874+
assert!(
7875+
first_string_end < full_len,
7876+
"second string must extend beyond the split point"
7877+
);
7878+
let short_buf = full_buf.slice_with_length(0, first_string_end);
7879+
let long_buf = full_buf.clone(); // full length — covers both strings
7880+
7881+
// Borrow the raw 128-bit view words from `base_array`. Each word
7882+
// encodes (length, prefix, buffer_index, offset); buffer_index is 0 in
7883+
// both because `base_array` has a single data buffer.
7884+
let views = base_array.views();
7885+
let view0_raw: u128 = views[0]; // first string — lives within short_buf
7886+
let view1_raw: u128 = views[1]; // second string — lives in long_buf only
7887+
7888+
let schema =
7889+
Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8View, true)]));
7890+
7891+
// Array 1: backed only by `short_buf` (shorter declared length)
7892+
let arr1: StringViewArray = {
7893+
let views_buf = ScalarBuffer::from(vec![view0_raw]);
7894+
// SAFETY: view0 was taken verbatim from `base_array` which is valid,
7895+
// and `short_buf` is sized to exactly cover the first string's byte
7896+
// range (offset 0..first_string_end), so all offsets referenced by
7897+
// view0 are in bounds.
7898+
unsafe {
7899+
GenericByteViewArray::new_unchecked(
7900+
views_buf,
7901+
vec![short_buf].into(),
7902+
None,
7903+
)
7904+
}
7905+
};
7906+
7907+
// Array 2: backed only by `long_buf` (full declared length)
7908+
let arr2: StringViewArray = {
7909+
let views_buf = ScalarBuffer::from(vec![view1_raw]);
7910+
// SAFETY: view1 was taken verbatim from `base_array` which is valid,
7911+
// and `long_buf` is the full allocation, so all byte ranges are covered.
7912+
unsafe {
7913+
GenericByteViewArray::new_unchecked(
7914+
views_buf,
7915+
vec![long_buf].into(),
7916+
None,
7917+
)
7918+
}
7919+
};
7920+
7921+
// Sanity-check the arrays read back correctly before we feed them in.
7922+
assert_eq!(
7923+
arr1.value(0),
7924+
"this is a long string that exceeds inline size 12"
7925+
);
7926+
assert_eq!(
7927+
arr2.value(0),
7928+
"another long string that exceeds inline size 12"
7929+
);
7930+
7931+
let batch1 = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(arr1)])?;
7932+
let batch2 = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(arr2)])?;
7933+
7934+
let metrics = BuildProbeJoinMetrics::new(0, &ExecutionPlanMetricsSet::new());
7935+
let pool: Arc<dyn MemoryPool> = Arc::new(UnboundedMemoryPool::default());
7936+
let batches = vec![batch1, batch2];
7937+
let (mut reservation, inputs_reserved) = reserve_inputs(&batches, &pool)?;
7938+
7939+
let batch = concat_build_batches(
7940+
&schema,
7941+
batches,
7942+
false,
7943+
inputs_reserved,
7944+
&mut reservation,
7945+
&metrics,
7946+
)?;
7947+
7948+
let view_arr = batch
7949+
.column(0)
7950+
.as_any()
7951+
.downcast_ref::<StringViewArray>()
7952+
.unwrap();
7953+
7954+
// Both rows must be readable after deduplication — the value that lives
7955+
// exclusively in the longer buffer must not have been dropped.
7956+
assert_eq!(
7957+
view_arr.value(0),
7958+
"this is a long string that exceeds inline size 12",
7959+
"value from the shorter-declared-length buffer should survive"
7960+
);
7961+
assert_eq!(
7962+
view_arr.value(1),
7963+
"another long string that exceeds inline size 12",
7964+
"value from the longer-declared-length buffer should survive"
7965+
);
7966+
assert_eq!(batch.num_rows(), 2);
7967+
Ok(())
7968+
}
7969+
77847970
/// The build side is concatenated into a single batch, and that copy must
77857971
/// be visible to the memory pool while the input batches are still alive.
77867972
#[rstest]

0 commit comments

Comments
 (0)