[query] Broadcast with array of array byte streams

[johnc1231]

I ran into issues when broadcasting a very large struct of ndarrays for a huge linear regression, where the the total size was more than MAX_INT bytes. To solve this, I've changed SerializableRegionValue to use ArrayOfByteArrayOutputStream and ArrayOfByteInputStream, which create nested arrays of bytes instead of just one array, removing any maximum length issues.

PRing now for tests, also running benchmarks.

[tpoterba]

Woah, so fast!

I think this is going to have poor performance until you implement the batched readers writers that look like

    public void write(byte b[], int off, int len) throws IOException
    public int read(byte b[], int off, int len) throws IOException

[johnc1231]

I agree that would be faster, but do we use those when we encode / decode broadcasted data? We should be if we aren't, but since it's working currently my suspicion is we aren't using them? Unless there's a default implementation where they work in terms of the single byte versions?

[johnc1231]

The gist suggests if anything things got slightly faster after doing this, though admittedly the sentinels all got faster too.: https://gist.github.com/johnc1231/cb10f22dad676bb30da680b9e2178614

[tpoterba]

This is really baffling. Maybe this code isn't really used in benchmarks in local mode? I would expect that every write/read from these buffers for SerializableRegionValue would go through the array/buffer read calls, because we use StreamBlockBuffer for the buffer spec:

  val wireSpec: BufferSpec = LEB128BufferSpec(
    BlockingBufferSpec(32 * 1024,
      LZ4SizeBasedBlockBufferSpec("fast", 32 * 1024,
        256,
        new StreamBlockBufferSpec)))

We won't call write/read on the output/input buffers except to write/read blocks.

[johnc1231]

Maybe it's because OutputStream.write has a default implementation in Java?

    public void write(byte b[], int off, int len) throws IOException {
        if (b == null) {
            throw new NullPointerException();
        } else if ((off < 0) || (off > b.length) || (len < 0) ||
                   ((off + len) > b.length) || ((off + len) < 0)) {
            throw new IndexOutOfBoundsException();
        } else if (len == 0) {
            return;
        }
        for (int i = 0 ; i < len ; i++) {
            write(b[off + i]);
        }
    }

So that comes baked in using the single byte write method

[tpoterba]

Maybe it's because OutputStream.write has a default implementation in Java?

Yeah, that default implementation should be WAY slower than an implementation that uses System.arrayCopy though. That's why I'm surprised you don't see performance regressions.

[johnc1231]

Ok, wrote the multi byte versions, running benchmarks.

[johnc1231]

https://gist.github.com/johnc1231/7667f228d022c636d42924dbd181bc96

Seems like not a lot of change, though it's definitely trending slower instead of trending faster the way the earlier one did.

[tpoterba]

one comment and ready to go

[tpoterba]

instead of having a currentArray index, can we just have a currentBuilder: ByteArrayOutputStream? I think having to do a lookup by index in the BoxedArrayBuilder in buffer might be a performance hit, and the code will probably be simpler regardless.

[johnc1231]

Addressed, want me to benchmark again?

Done

[tpoterba]

nah, ship it.