andygrove commented on code in PR #7016:
URL: https://github.com/apache/datafusion-comet/pull/7016#discussion_r4338896991
##########
spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala:
##########
@@ -341,10 +341,65 @@ object Utils extends CometTypeShim with Logging {
// use Spark's compression codec (LZ4 by default) and not Comet's
compression
val codec = CompressionCodec.createCodec(SparkEnv.get.conf)
- val cbbis = bytes.toInputStream()
- val ins = new DataInputStream(codec.compressedInputStream(cbbis))
// batches are in Arrow IPC format
- new ArrowReaderIterator(Channels.newChannel(ins), source)
+ new ArrowReaderIterator(ipcChannel(bytes, codec), source)
+ }
+
+ /**
+ * A channel over the Arrow IPC stream in `bytes`, which
[[serializeBatches]] wrote through
+ * `codec`.
+ */
+ def ipcChannel(bytes: ChunkedByteBuffer, codec: CompressionCodec):
ReadableByteChannel =
+ new InputStreamChannel(codec.compressedInputStream(bytes.toInputStream()))
+
+ /** Largest piece that [[InputStreamChannel]] reads from its stream at a
time. */
+ private[util] val MaxReadPiece = 64 * 1024
+
+ /**
+ * Reads `in` into each buffer it is asked to fill until the buffer is full
or `in` ends, up to
+ * [[MaxReadPiece]] bytes at a time through one reused array.
`Channels.newChannel` reads 8 KiB
+ * at a time, and enters and leaves the JDK's interruptible I/O machinery
for each piece. Like
+ * that channel, this one closes and throws `ClosedByInterruptException`
before it reads a piece
+ * on an interrupted thread, so a cancelled decode stops within a piece of
where it was.
+ */
+ private final class InputStreamChannel(in: InputStream) extends
ReadableByteChannel {
Review Comment:
That's fair. The one timing doesn't justify the extra code, and it did cost
a second commit to match the JDK's interrupt behavior. I've dropped it from
this PR, so the cache decodes through `Channels.newChannel` and turning the
setting off gives the `branch-1.1` decode path back. If a per-task decode
benchmark with sharing off shows it matters, it can come back as its own PR.
--
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]
