ReadData and DataBlockCodec

[tpietzsch]

Add DataBlock methods
void readData(ByteOrder, ReadData) and
void writeData(ByteOrder, OutputStream)
remove
ByteBuffer toByteBuffer() and
void readData(ByteBuffer).

Add Splittable.ReadData, etc.

Still not fully settled...

[cmhulbert]

I think there is some overlap with this branch, I'll be back at work on Monday fully, so hopefully we can distill the important parts from these two. My branch is more focused on getting this working for sharding/codecs, but hopefully there is a lot of shared logic. I created a draft PR to better compare at #138

Some specific files that are likely to overlap:
SplitableData.java
SplitKeyValueAccessData.java
SplitByteBufferedData.java

[tpietzsch]

I'm still working on this, so please wait for a bit (just until later today, probably) before starting to reconcile.

In this PR, the ReadData is turning out more of an abstraction over byte[], ByteBuffer, InputStream, OutputStream. I'm changing the API to just always hand around ReadData. I'll write more about it when it has settled.
So that there is also SplittableReadData (extending ReadData) is not the important thing. It's more about cleaning up the DataBlock/Compression/BlockWriter/BlockReader API.

Probably we should set up a zoom meeting to discuss, later this week

[cmhulbert]

Sounds good to me, thanks for taking a stab at this. I'll clean up some of what I have, and let me know when you have time to chat, would be happy to zoom

[tpietzsch]

related PRs:

JaneliaSciComp/n5-zstandard#4
saalfeldlab/n5-blosc#11
saalfeldlab/n5-zarr#63

[tpietzsch]

Ok, I'm happy now.

The ReadData abstraction is used throughout, now:

DataBlock de/serialization:

public interface DataBlock<T> {
    void readData(ReadData readData) throws IOException;
    ReadData writeData(ByteOrder byteOrder);
}

BytesCodec/Compression interface:

public interface BytesCodec {
    ReadData decode(ReadData readData, int decodedLength) throws IOException;
    ReadData encode(ReadData readData) throws IOException;
}

block reader:

    final ReadData data = ReadData.from(in)
                .decode(datasetAttributes.getCompression(), numBytes);
    dataBlock.readData(data);

and block writer:

    dataBlock.writeData(ByteOrder.BIG_ENDIAN)
                .encode(datasetAttributes.getCompression())
                .writeTo(out);

There is also extensive javadoc for ReadData, BytesCodec.

@axtimwalde @bogovicj @cmhulbert Let's set up a review session this week? I think this is ready to merge.

[bogovicj]

A few small comments / observations

[bogovicj]

I'd prefer a different name for this. Alternatively, as we discussed maybe if we add a DataBlockCodec interface, this method may not even be needed.

• serialize
• encode
  • this reflects that DataBlocks are currently responsible for the encoding that will later be done by a codec

[bogovicj]

is this needed?

[tpietzsch]

oops. no. that's left over from debugging....

[bogovicj]

Is it possible / do we want to be able to split at some offset and have the length go "to the end" of the ReadData?

maybe length = -1? (but that's forbidden in some places) or length = Long.MAX_VALUE ?

[tpietzsch]

I don't know, because so far I didn't need splitting at all, but it sounds like that could be useful!

If the ReadData knows its length, then you can just data.split(offset, data.length() - offset).

Split "to the end" would be helpful for ReadData that don't know their length (and to avoid having to retrieve it).
Maybe we just add ReadData split(long offset) without a length argument to split to the end.

At the moment, the javadoc of ReadData.splittable() says

The returned SplittableReadData has a known length and multiple inputStreams can be opened on it.

That contract should be changed then, probably.
Maybe we should add an explicit ReadData.materialize() method for that instead.

[bogovicj]

Something related I came across while experimenting. How would you feel about a method limit(long)?
It should be equivalent to split(0,long), when splittable, but could be useful in other cases, specifically:

final InputStream ones = new InputStream() {
     @Override public int read() throws IOException { return 1; }
};

// This fails because `splittable` tries to read the whole (infinite) input stream which is reasonable.
ReadData firstTenBySpllit = ReadData.from(ones).splittable().split(0, 10);

// this could be a sensible alternative
ReadData firstTen = ReadData.from(ones).limit(10);

[tpietzsch]

That is more difficult, I think.
Ideally ReadData should be considered immutable.
Conceptually, ReadData.split(...) shouldn't modify the ReadData that you call it on.

However

final InputStream sequence = new InputStream() {
	private int value = 0;
	@Override public int read() throws IOException { return value++; }
};

ReadData data = ReadData.from(sequence);
// data.inputStream().read() would be 0 here

ReadData firstTen = data.limit(10);
// data.inputStream().read() would be 10 here

Therefore, a limit() implementation that reads a bit of the InputStream turns data unusable.

Maybe then, limit(n) should be considered destructive and return a pair of ReadData, one for the first n bytes, and one for the rest.

That would be an actual "split" then.
And maybe we should re-consider naming...
The current split() is actually more a "view" than a "split". Maybe slice(...) would be relatively consistent with ByteBuffer terminology?

[tpietzsch]

... but I totally agree that limit() seems useful!

[bogovicj]

Ah, I see what you mean, good point.

Maybe then, limit(n) should be considered destructive and return a pair of ReadData, one for the first n bytes, and one for the rest.

The current split() is actually more a "view" than a "split". Maybe slice(...) would be relatively consistent with ByteBuffer terminology?

This makes sense to me - I'd be happy with something like:

final InputStream sequence = new InputStream() {
	private int value = 0;
	@Override public int read() throws IOException { return value++; }
};

ReadData data = ReadData.from(sequence);

SplitReadData split = data.split(10);  // basically a proposed "limit", a Pair<ReadData,ReadData>

ReadData firstTen = split.first();
ReadData lastInfinity = split.last();
ReadData view = firstTen.slice(5,5); // what is currently called "split"

[tpietzsch]

I separated de/serialization out of DataBlock<T> completely, into DataBlockCodec<T>. (I had started to do the "incremental" version that we had discussed, but in the end just did the whole thing.)
The DataBlock implementations are super basic now, e.g.

public class LongArrayDataBlock extends AbstractDataBlock<long[]> {
	public LongArrayDataBlock(final int[] size, final long[] gridPosition, final long[] data) {
		super(size, gridPosition, data, a -> a.length);
	}
}

public class StringDataBlock extends AbstractDataBlock<String[]> {
	public StringDataBlock(final int[] size, final long[] gridPosition, final String[] data) {
		super(size, gridPosition, data, a -> a.length);
	}
}

The DataBlockCodec interface looks like this:

public interface DataBlockCodec<T> {
	ReadData encode(DataBlock<T> dataBlock) throws IOException;
	DataBlock<T> decode(ReadData readData, long[] gridPosition) throws IOException;
}

That looks like it could be quite compatible with @bogovicj DataBlockInputStream ideas?
DataBlockCodec can be obtained from DatasetAttributes. (Which is very nice, because ZarrDatasetAttributes can override it, and consequently we can just use DefaultBlockReader/Writer for Zarr.)

As discussed in the last code review, I also did the following:

• I removed ByteOrder from ReadData again.
• I replaced the OutputStreamEncoder interface (bundling OutputStream and finish method) with a ProxyOutputStream wrapper that blocks the close() method. I couldn't use UnaryOperator<OutputStream> though, because I want to throw IOException.
• I removed the ReadData.encode/decode methods that just called return Compression.encode/decode(this).
• I inlined the BytesCodec interface into Compression (for now).

[tpietzsch]

I moved the SplittableReadData part to a separate PR #140, so that this one can be merged.
(The "splittable" part was unused and should only be relevant to Zarr v3 / sharding ...)

@bogovicj @cmhulbert @axtimwalde Can we merge this already?
We could also set up another code review for next week to look at the datails.

[cmhulbert]

Thanks @tpietzsch! I'll plan to spend some time today and tomorrow to try and rebase my current branch off this, and see how it plays with the codecs, especially when chaining arbitrary codecs. That should hopefully expose any thing else we need. Would also be happy to chat some time next week

[bogovicj]

A few small ideas about names.

One bigger question about how LazyReadData works

Edit:
I may be slowly starting to understand LazyReadData

The point is that we don't have an output stream where the encoders are defined - their job is to create a ReadData from something else, and that ReadData can write itself to whatever output stream it is given.

[bogovicj]

I'd call this BlockHeader since everything in N5 uses "block" not "chunk"

[tpietzsch]

done

[bogovicj]

maybe rename to ByteArrayReadData since we're not doing the splittable part yet?
I'm also happy to keep the name as is since the splitting will come soon.

[tpietzsch]

I'd leave it as is. This class is anyway package-private, so it is easy to rename later if necessary

[bogovicj]

out.flush() is called in all the encode methods here - should it be the responsibility of the OutputStreamOperator to flush? could it instead be the job of LazyReadData?

I'll make a note there too.

Edit: okay maybe I get it now. OutputStreamWriter is a functional interface and can't be flushed itself.
This leads to a different question.

[tpietzsch]

That's a good point, actually. I'll have another look at who should be responsible for flush().

[tpietzsch]

I revised where flush() happens. 👍

[bogovicj]

I'm confused here. What's the benefit of using this functional interface instead of an OutputStream directly?

Edit: I may get the point - comment above. (sorry for the spread out comments)

[tpietzsch]

The OutputStreamWriter is a way to defer the writing/generation/serialization (not sure what to best call it) of the data until it is needed and it is clear what the destination is.

The Compression implementations (most of them) use Input/OutputStream wrappers to do the compression. The DataBlockCodec::encode methods use these Compressions so they have to use OutputStream too. Because they don't take an OutputStream argument we cannot serialize into an OutputStream directly.

Instead DataBlockCodec::encode returns a LazyReadData that is not committed to any OutputStream yet --

• Maybe someone needs serialized data as a byte[] array. In that case, the LazyReadData will create a ByteArrayOutputStream and use the OutputStreamWriter to populate it.
• Maybe someone wants to write the serialized data into a FileOutputStream. In that case, the LazyReadData can pass that FileOutputStream to the OutputStreamWriter directly (without intermediate serialization into a byte[] array).