andygrove opened a new issue, #1424:
URL: https://github.com/apache/arrow-java/issues/1424
### Describe the bug, including details regarding any error messages,
version, and platform.
`VectorUnloader` (and so `ArrowStreamWriter` / `ArrowFileWriter`) serializes
a variable-width or list vector by writing its data buffer from byte 0 up to
the last offset, and a list's child vector from element 0. When the vector's
offsets do not start at zero, everything before the first offset is written
into the record batch body even though no row references it.
Offsets that start past zero are valid in the columnar format, and they are
what the C Data interface produces for a slice: arrow-rs slices a
`StringArray`, `BinaryArray`, `ListArray` or `MapArray` by advancing the start
of its offsets buffer, and `BufferImportTypeVisitor` keeps the offsets as
exported and sizes the data buffer to the last offset
(`BufferImportTypeVisitor.visit(ArrowType.Utf8)` and friends). So a batch that
is the k-th slice of a larger native array is unloaded with the bytes of all
k-1 slices before it. In DataFusion Comet, where a grouped aggregate emits its
output in `batchSize`-row slices of one batch, a partition that emits K slices
caches about (K+1)/2 times its variable-width data, and the inflated sizes feed
Spark's plan statistics (apache/datafusion-comet#6995,
apache/datafusion-comet#6779).
The other writers re-base in this case: arrow-rs's IPC writer re-encodes
offsets to start at zero and only writes the referenced value bytes
(`reencode_offsets` in `arrow-ipc/src/writer.rs`), and the C++
`RecordBatchSerializer` does the same (`GetZeroBasedValueOffsets`). Java's
`VectorAppender` was fixed to handle a non-zero start offset in #1180
(GH-1179); `VectorUnloader` is the remaining consumer that assumes offsets
start at zero.
Reproduced on `main` (67ef5abf3, 20.0.0-SNAPSHOT) with JDK 17 on macOS; the
buffer sizing in `VectorUnloader` is unchanged since 18.3.0, which is what
Comet runs. The test below builds the vectors a C Data import of a slice would
produce (rows 2..3 of `["aa","bb","cc","dd"]` and rows 1..2 of
`[[0,1],[2,3],[4,5],[6,7]]`) and unloads them:
```java
BufferAllocator a = new RootAllocator();
VarCharVector s = new VarCharVector("s", a);
s.loadFieldBuffers(new ArrowFieldNode(2, 0),
Arrays.asList(allValid(a), ints(a, 4, 6, 8), bytes(a, "aabbccdd")));
// data sized to last offset, as the import does
ListVector l = ListVector.empty("l", a);
l.addOrGetVector(FieldType.nullable(MinorType.INT.getType()));
l.loadFieldBuffers(new ArrowFieldNode(2, 0), Arrays.asList(allValid(a),
ints(a, 2, 4, 6)));
((IntVector) l.getDataVector()).loadFieldBuffers(new ArrowFieldNode(8, 0),
Arrays.asList(allValid(a), ints(a, 0, 1, 2, 3, 4, 5, 6, 7)));
VectorSchemaRoot root = new VectorSchemaRoot(Arrays.asList(s, l));
root.setRowCount(2);
// s = [cc, dd], l = [[2,3],[4,5]]
try (ArrowRecordBatch batch = new VectorUnloader(root).getRecordBatch()) {
System.out.println(batch.getNodes());
System.out.println(batch.getBuffersLayout());
}
try (VectorSchemaRoot sliced = root.slice(0, 2);
ArrowRecordBatch batch = new VectorUnloader(sliced).getRecordBatch()) {
System.out.println(batch.getNodes());
}
```
Output:
```
[ArrowFieldNode [length=2], ArrowFieldNode [length=2], ArrowFieldNode
[length=6]]
[validity size=1, offsets size=12, data size=8, validity size=1, offsets
size=12, child validity size=1, child data size=24]
[ArrowFieldNode [length=2], ArrowFieldNode [length=2], ArrowFieldNode
[length=4]]
```
The string data buffer is written as 8 bytes (`aabbccdd`) where the two rows
reference 4 (`ccdd`), and the list child is written as 6 elements (24 bytes)
where the two lists reference 4. The stream reads back correctly, with the
offsets still starting at 4 and 2, so this is a size problem rather than a
correctness one, but it compounds with every slice. `VectorSchemaRoot.slice(0,
rowCount)` produces the expected node lengths, because `splitAndTransfer`
re-bases offsets, and that is the workaround Comet applies before every
`VectorUnloader` call (`Utils.withRebasedOffsets`).
**Expected:** `VectorUnloader` writes only the bytes (and, for lists and
maps, the child elements) between the first and last offset, re-basing the
offsets to start at zero, as the Rust and C++ IPC writers do. Alternatively,
the C Data import could re-base at import time, which would fix every consumer
at once at the cost of an offsets copy on every sliced import.
--
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]
