mbutrovich commented on code in PR #6191:
URL: https://github.com/apache/datafusion-comet/pull/6191#discussion_r4095031420
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -727,9 +727,9 @@ object CometConf extends ShimCometConf {
"shuffle data to disk. Larger values may improve write performance by
reducing " +
"the number of system calls, but will use more memory. " +
"The default is 1MB which provides a good balance between performance
and memory usage.")
- .bytesConf(ByteUnit.MiB)
+ .bytesConf(ByteUnit.BYTE)
.checkValue(v => v > 0, "Write buffer size must be positive")
- .createWithDefault(1)
+ .createWithDefault(1024 * 1024)
Review Comment:
With this default taking effect, how much untracked memory does a
shuffle-writing task hold? I count four buffers of this size in the
multi-partition local writer once it has spilled, not three. They are the data
file `BufWriter` ([`local_partition_writer.rs`
L127](https://github.com/apache/datafusion-comet/blob/e74d597b2bab4320143c969d9c1d20d3133c6479/native/shuffle/src/writers/local/local_partition_writer.rs#L127)),
the spill file `BufWriter`, which stays open until the task ends ([`spill.rs`
L222](https://github.com/apache/datafusion-comet/blob/e74d597b2bab4320143c969d9c1d20d3133c6479/native/shuffle/src/writers/local/spill.rs#L222)),
the spill copy buffer ([`local_partition_writer.rs`
L286](https://github.com/apache/datafusion-comet/blob/e74d597b2bab4320143c969d9c1d20d3133c6479/native/shuffle/src/writers/local/local_partition_writer.rs#L286)),
and the recycled scratch, which `flush` shrinks back to this size rather than
to zero ([`buf_batch_writer.rs` L210-L212](https://github.
com/apache/datafusion-comet/blob/e74d597b2bab4320143c969d9c1d20d3133c6479/native/shuffle/src/writers/buf_batch_writer.rs#L210-L212)).
With the 1-byte buffer all four were close to zero, so this PR moves about 4
MiB per task from nothing into untracked native memory, and more if a user
raises the setting.
Could the writer reserve these through the pool, for example by growing the
`ShuffleRepartitioner` reservation by the buffer sizes when it creates them? If
you'd rather keep that out of this PR, could you file a tracking issue for it
and link it here? Either way, the PR description should say 4 MiB rather than 3.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -727,9 +727,9 @@ object CometConf extends ShimCometConf {
"shuffle data to disk. Larger values may improve write performance by
reducing " +
"the number of system calls, but will use more memory. " +
"The default is 1MB which provides a good balance between performance
and memory usage.")
- .bytesConf(ByteUnit.MiB)
+ .bytesConf(ByteUnit.BYTE)
.checkValue(v => v > 0, "Write buffer size must be positive")
Review Comment:
`CometNativeShuffleWriter` clamps this value with `.min(Int.MaxValue).toInt`
because the proto field is `int32` ([`CometNativeShuffleWriter.scala`
L326-L327](https://github.com/apache/datafusion-comet/blob/e74d597b2bab4320143c969d9c1d20d3133c6479/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala#L326-L327)).
Now that the value is in bytes, `3g` passes this check and is silently
replaced by 2 GiB, which is the kind of silent substitution the rest of this PR
removes. How about rejecting it here, so the writer can use a plain `.toInt`?
```suggestion
.checkValue(
v => v > 0 && v <= Int.MaxValue,
s"Write buffer size must be between 1 and ${Int.MaxValue} 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]
