andygrove commented on code in PR #6861:
URL: https://github.com/apache/datafusion-comet/pull/6861#discussion_r4241898969
##########
native/shuffle/src/spark_unsafe/row.rs:
##########
@@ -1423,8 +1440,17 @@ pub fn process_sorted_row_partition(
frozen.clear();
let mut cursor = Cursor::new(&mut frozen);
- let block_writer =
ShuffleBlockWriter::try_new(batch.schema().as_ref(), codec.clone())?;
- written += block_writer.write_batch(&batch, &mut cursor, &mut
codec_context, &ipc_time)?;
+ // The schema changes between batches when a dictionary-encoded column
falls back to a
+ // plain array (see `builder_to_array`).
+ if block_writer
Review Comment:
Nit, optional: the cache invalidation here has no test. If I replace the
condition with `block_writer.is_none()`, all 172 unit tests in the shuffle
crate still pass. Without the schema check, a column that goes from plain to
dictionary fails the write with `Ipc error: no dict id for field`. One that
goes from dictionary to plain writes a block that decodes to the wrong type. I
wrote a throwaway test that calls `process_sorted_row_partition` with
`batch_size = 4` and a dictionary ratio of 1.5 over `x,x,x,x,a,b,c,d,y,y,y,y`.
The column goes dictionary, plain, dictionary inside one call, and the PR
decodes all three blocks correctly. Could we add a test like that? It needs
only a few one-column `UnsafeRow`s, laid out the way `benches/row_columnar.rs`
does.
##########
native/spark-expr/src/string_funcs/split.rs:
##########
@@ -445,18 +445,24 @@ fn push_split_parts<O: OffsetSizeTrait>(
builder: &mut GenericStringBuilder<O>,
) {
if limit == 0 {
Review Comment:
Nit, optional: nothing in the repo would notice if this branch were wrong.
`test_split_limit_zero` goes through `split_string`, which calls
`split_with_regex`, and this PR doesn't touch that helper. The only other
`limit = 0` test I found splits `a,b,c,d,e`, which has no empty parts. I broke
the branch three ways (kept the trailing empties, dropped the interior ones,
removed the `[""]` fallback) and all 16 split tests still passed. The logic
itself looks right to me. I compared it with the old branch on every string up
to length 8 over a three-character alphabet with a handful of patterns, about
88k cases, and they matched. Could we add a test through `spark_split` with
limit 0 on `a,b,c,,`, `a,,b`, `,a`, `a,,b,,` and the empty string? A
`split_rust.sql` query with the same inputs would cover it end to end. I'd
leave `,,` out of the SQL file since Java returns an empty array for it and the
native path returns `[""]`.
##########
native/shuffle/src/spark_unsafe/row.rs:
##########
@@ -517,9 +521,12 @@ fn append_nested_struct_fields_field_major(
let nested_builder = get_field_builder!(struct_builder,
StructBuilder, field_idx);
// Collect nested struct addresses and sizes in one pass,
building validity
- let mut nested_addresses: Vec<jlong> =
Vec::with_capacity(num_rows);
- let mut nested_sizes: Vec<jint> = Vec::with_capacity(num_rows);
- let mut nested_is_null: Vec<bool> =
Vec::with_capacity(num_rows);
+ nested_addresses.clear();
Review Comment:
Nit, optional: same question for the scratch vectors here and at line 981.
Reusing them only matters when a struct has two or more nested struct fields,
and I didn't see a case with that shape in `CometColumnarShuffleSuite`. If I
remove the three `clear()` calls in both functions, the shuffle crate's tests
still pass. A struct with `a: struct<x>`, `b: struct<y>` and `m: struct<c:
struct<p>, d: struct<q>>`, with a null at each level, passes on the PR and
fails without the `clear()` calls. Could the test above also cover that shape?
##########
native/spark-expr/src/bloom_filter/spark_bloom_filter.rs:
##########
@@ -157,19 +156,19 @@ impl SparkBloomFilter {
/// Serializes a SparkBloomFilter to a byte array conforming to Spark's
BloomFilter
/// binary format. The output format follows the filter's `version`.
pub fn spark_serialization(&self) -> Vec<u8> {
- let mut out: Vec<u8> = (self.version.to_int() as
u32).to_be_bytes().to_vec();
- out.append(&mut self.num_hash_functions.to_be_bytes().to_vec());
+ // At most a 16-byte header (version, numHashFunctions, seed,
numWords), then the words.
+ let mut out: Vec<u8> = Vec::with_capacity(16 + self.bits.word_size() *
8);
+ out.extend_from_slice(&(self.version.to_int() as u32).to_be_bytes());
+ out.extend_from_slice(&self.num_hash_functions.to_be_bytes());
if let SparkBloomFilterVersion::V2 = self.version {
// Spark's BloomFilterImplV2.writeTo writes the seed between
// numHashFunctions and the bit array.
- out.append(&mut (self.seed as u32).to_be_bytes().to_vec());
+ out.extend_from_slice(&(self.seed as u32).to_be_bytes());
}
- out.append(&mut (self.bits.word_size() as u32).to_be_bytes().to_vec());
- let mut filter_state: Vec<u64> = self.bits.data();
- for i in filter_state.iter_mut() {
- *i = i.to_be();
+ out.extend_from_slice(&(self.bits.word_size() as u32).to_be_bytes());
+ for word in self.bits.data() {
Review Comment:
Nit, optional: in a local A/B against the old code, this per-word
`extend_from_slice` loop ran at about the same speed. It was a few microseconds
either way from 64 KiB to 1 MiB, and the sign flipped between runs at 8 MiB.
Peak memory is 3x lower, which is the real win. Filling the presized buffer
with `resize` and `chunks_exact_mut(8)` ran about 2.5x faster than this loop at
1 MiB, roughly 18 µs against 50 µs, with one allocation and byte-identical
output. Could we use that form?
--
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]
