jbonofre commented on code in PR #1364:
URL: https://github.com/apache/arrow-java/pull/1364#discussion_r4295213565
##########
vector/src/main/java/org/apache/arrow/vector/util/VectorAppender.java:
##########
@@ -153,10 +153,9 @@ public ValueVector visit(BaseVariableWidthVector
deltaVector, Void value) {
targetVector.getValidityBuffer());
// append data buffer
- MemoryUtil.copyMemory(
- deltaVector.getDataBuffer().memoryAddress() + deltaDataStart,
- targetVector.getDataBuffer().memoryAddress() + targetDataSize,
- deltaDataSize);
+ targetVector
Review Comment:
Nice improvement, this turns a raw out-of-bounds native copy into a checked
one. One concern is ordering. By the time this `setBytes` runs, the target's
data buffer has been reallocated for `deltaDataSize` and `concatBits` has
already written the delta's validity bits into the target.
If the check throws, the target is left with extra validity bits while
`valueCount`, offsets and `lastSet` are unchanged.
Could we validate the delta (last offset against its data buffer capacity)
before any realloc or write, so a rejected delta leaves the target untouched?
##########
vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowReaderWriter.java:
##########
@@ -649,6 +649,66 @@ public void testDeltaDictionary() throws Exception {
AutoCloseables.close(closeableList);
}
+ @Test
+ public void testDeltaDictionaryOffsetsExceedDataBuffer() throws Exception {
+ try (VarCharVector dictVector = newVarCharVector("dict", allocator)) {
+ setVector(
+ dictVector,
+ "foo".getBytes(StandardCharsets.UTF_8),
+ "bar".getBytes(StandardCharsets.UTF_8));
+ Dictionary dictionary = new Dictionary(dictVector, new
DictionaryEncoding(1L, false, null));
+ DictionaryProvider.MapDictionaryProvider provider =
+ new DictionaryProvider.MapDictionaryProvider();
+ provider.put(dictionary);
+ Field encodedField =
+ new Field(
+ "f",
+ new FieldType(true, new ArrowType.Int(32, true),
dictionary.getEncoding()),
+ null);
+ Schema schema =
+ new Schema(
+ Collections.singletonList(
+ DictionaryUtility.toMessageFormat(encodedField, provider,
new HashSet<>())));
+
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ WriteChannel out = new WriteChannel(newChannel(outStream));
+ MessageSerializer.serialize(out, schema);
+
+ List<AutoCloseable> closeableList = new ArrayList<>();
+ serializeDictionaryBatch(out, dictionary, false, closeableList);
+
+ // delta dictionary whose offsets point far past its 8-byte data buffer
+ ArrowBuf validity = allocator.buffer(0);
+ ArrowBuf offsets = allocator.buffer(12);
+ offsets.setInt(0, 0);
+ offsets.setInt(4, 4);
+ offsets.setInt(8, 4096);
+ offsets.writerIndex(12);
+ ArrowBuf data = allocator.buffer(8);
+ data.setBytes(0, "bazzquxx".getBytes(StandardCharsets.UTF_8));
+ data.writerIndex(8);
+ ArrowRecordBatch deltaBatch =
+ new ArrowRecordBatch(
+ 2,
+ Collections.singletonList(new ArrowFieldNode(2, 0)),
+ Arrays.asList(validity, offsets, data));
+ ArrowDictionaryBatch deltaDictionaryBatch = new ArrowDictionaryBatch(1L,
deltaBatch, true);
+ MessageSerializer.serialize(out, deltaDictionaryBatch);
+ closeableList.add(deltaDictionaryBatch);
+ AutoCloseables.close(validity, offsets, data);
+
+ // write eos
+ out.writeIntLittleEndian(0);
+
+ try (ArrowStreamReader reader =
+ new ArrowStreamReader(
+ new
ByteArrayReadableSeekableByteChannel(outStream.toByteArray()), allocator)) {
+ assertThrows(IndexOutOfBoundsException.class, reader::loadNextBatch);
+ }
+ AutoCloseables.close(closeableList);
Review Comment:
If `assertThrows` fails, `closeableList` is never closed and the allocator
leaks. Consider wrapping it in a `finally` so the original assertion failure
isn't masked by a leak error.
##########
vector/src/main/java/org/apache/arrow/vector/util/VectorAppender.java:
##########
@@ -229,10 +228,9 @@ public ValueVector visit(BaseLargeVariableWidthVector
deltaVector, Void value) {
targetVector.getValidityBuffer());
// append data buffer
- MemoryUtil.copyMemory(
- deltaVector.getDataBuffer().memoryAddress() + deltaDataStart,
- targetVector.getDataBuffer().memoryAddress() + targetDataSize,
- deltaDataSize);
+ targetVector
Review Comment:
The same comments applies here 😄
##########
vector/src/main/java/org/apache/arrow/vector/util/VectorAppender.java:
##########
@@ -153,10 +153,9 @@ public ValueVector visit(BaseVariableWidthVector
deltaVector, Void value) {
targetVector.getValidityBuffer());
// append data buffer
- MemoryUtil.copyMemory(
- deltaVector.getDataBuffer().memoryAddress() + deltaDataStart,
- targetVector.getDataBuffer().memoryAddress() + targetDataSize,
- deltaDataSize);
+ targetVector
+ .getDataBuffer()
+ .setBytes(targetDataSize, deltaVector.getDataBuffer(), deltaDataStart,
deltaDataSize);
Review Comment:
`deltaDataSize` is derived from the delta's offsets without validation. A
last offset smaller than the first gives a negative size, and a very large one
can overflow the int when added to `targetDataSize`.
Intermediate offsets are not checked against the data size either, so later
reads of the appender vector could go out of bounds even when the final offset
passes.
It would be good to guard `0 <= start <= end <= dataBuffer.capacity()` and
use `Math.addExact` or similar.
##########
vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowReaderWriter.java:
##########
@@ -649,6 +649,66 @@ public void testDeltaDictionary() throws Exception {
AutoCloseables.close(closeableList);
}
+ @Test
+ public void testDeltaDictionaryOffsetsExceedDataBuffer() throws Exception {
+ try (VarCharVector dictVector = newVarCharVector("dict", allocator)) {
+ setVector(
+ dictVector,
+ "foo".getBytes(StandardCharsets.UTF_8),
+ "bar".getBytes(StandardCharsets.UTF_8));
+ Dictionary dictionary = new Dictionary(dictVector, new
DictionaryEncoding(1L, false, null));
+ DictionaryProvider.MapDictionaryProvider provider =
+ new DictionaryProvider.MapDictionaryProvider();
+ provider.put(dictionary);
+ Field encodedField =
+ new Field(
+ "f",
+ new FieldType(true, new ArrowType.Int(32, true),
dictionary.getEncoding()),
+ null);
+ Schema schema =
+ new Schema(
+ Collections.singletonList(
+ DictionaryUtility.toMessageFormat(encodedField, provider,
new HashSet<>())));
+
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ WriteChannel out = new WriteChannel(newChannel(outStream));
+ MessageSerializer.serialize(out, schema);
+
+ List<AutoCloseable> closeableList = new ArrayList<>();
+ serializeDictionaryBatch(out, dictionary, false, closeableList);
+
+ // delta dictionary whose offsets point far past its 8-byte data buffer
+ ArrowBuf validity = allocator.buffer(0);
+ ArrowBuf offsets = allocator.buffer(12);
+ offsets.setInt(0, 0);
+ offsets.setInt(4, 4);
+ offsets.setInt(8, 4096);
+ offsets.writerIndex(12);
+ ArrowBuf data = allocator.buffer(8);
+ data.setBytes(0, "bazzquxx".getBytes(StandardCharsets.UTF_8));
+ data.writerIndex(8);
+ ArrowRecordBatch deltaBatch =
+ new ArrowRecordBatch(
+ 2,
+ Collections.singletonList(new ArrowFieldNode(2, 0)),
+ Arrays.asList(validity, offsets, data));
+ ArrowDictionaryBatch deltaDictionaryBatch = new ArrowDictionaryBatch(1L,
deltaBatch, true);
+ MessageSerializer.serialize(out, deltaDictionaryBatch);
+ closeableList.add(deltaDictionaryBatch);
+ AutoCloseables.close(validity, offsets, data);
+
+ // write eos
+ out.writeIntLittleEndian(0);
+
+ try (ArrowStreamReader reader =
+ new ArrowStreamReader(
+ new
ByteArrayReadableSeekableByteChannel(outStream.toByteArray()), allocator)) {
Review Comment:
This only covers the `VarChar` path.
Could we add a `LargeVarChar` case for the `BaseLargeVariableWidthVector`
visit? A positive case with a valid delta whose data start offset is non-zero
would also help, since that exercises the `deltaDataStart` argument to
`setBytes`.
##########
vector/src/test/java/org/apache/arrow/vector/ipc/TestArrowReaderWriter.java:
##########
@@ -649,6 +649,66 @@ public void testDeltaDictionary() throws Exception {
AutoCloseables.close(closeableList);
}
+ @Test
+ public void testDeltaDictionaryOffsetsExceedDataBuffer() throws Exception {
+ try (VarCharVector dictVector = newVarCharVector("dict", allocator)) {
+ setVector(
+ dictVector,
+ "foo".getBytes(StandardCharsets.UTF_8),
+ "bar".getBytes(StandardCharsets.UTF_8));
+ Dictionary dictionary = new Dictionary(dictVector, new
DictionaryEncoding(1L, false, null));
+ DictionaryProvider.MapDictionaryProvider provider =
+ new DictionaryProvider.MapDictionaryProvider();
+ provider.put(dictionary);
+ Field encodedField =
+ new Field(
+ "f",
+ new FieldType(true, new ArrowType.Int(32, true),
dictionary.getEncoding()),
+ null);
+ Schema schema =
+ new Schema(
+ Collections.singletonList(
+ DictionaryUtility.toMessageFormat(encodedField, provider,
new HashSet<>())));
+
+ ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+ WriteChannel out = new WriteChannel(newChannel(outStream));
+ MessageSerializer.serialize(out, schema);
+
+ List<AutoCloseable> closeableList = new ArrayList<>();
+ serializeDictionaryBatch(out, dictionary, false, closeableList);
+
+ // delta dictionary whose offsets point far past its 8-byte data buffer
+ ArrowBuf validity = allocator.buffer(0);
+ ArrowBuf offsets = allocator.buffer(12);
+ offsets.setInt(0, 0);
+ offsets.setInt(4, 4);
+ offsets.setInt(8, 4096);
+ offsets.writerIndex(12);
+ ArrowBuf data = allocator.buffer(8);
+ data.setBytes(0, "bazzquxx".getBytes(StandardCharsets.UTF_8));
+ data.writerIndex(8);
+ ArrowRecordBatch deltaBatch =
+ new ArrowRecordBatch(
+ 2,
+ Collections.singletonList(new ArrowFieldNode(2, 0)),
+ Arrays.asList(validity, offsets, data));
+ ArrowDictionaryBatch deltaDictionaryBatch = new ArrowDictionaryBatch(1L,
deltaBatch, true);
+ MessageSerializer.serialize(out, deltaDictionaryBatch);
+ closeableList.add(deltaDictionaryBatch);
+ AutoCloseables.close(validity, offsets, data);
+
+ // write eos
+ out.writeIntLittleEndian(0);
+
+ try (ArrowStreamReader reader =
+ new ArrowStreamReader(
+ new
ByteArrayReadableSeekableByteChannel(outStream.toByteArray()), allocator)) {
Review Comment:
`setBytes` bounds checking is gated by
`BoundsChecking.BOUNDS_CHECKING_ENABLED`. With
`arrow.enable_unsafe_memory_access=true` the check is skipped, no exception is
thrown, and this test fails (and the fix does nothing in that mode). Consider
skipping the test with an `assumeTrue` when bounds checking is off. Also,
asserting only the exception type lets the partial-mutation issue go unnoticed.
After the failure, please assert that the dictionary vector still has its
original two values.
##########
vector/src/main/java/org/apache/arrow/vector/util/VectorAppender.java:
##########
@@ -153,10 +153,9 @@ public ValueVector visit(BaseVariableWidthVector
deltaVector, Void value) {
targetVector.getValidityBuffer());
// append data buffer
- MemoryUtil.copyMemory(
- deltaVector.getDataBuffer().memoryAddress() + deltaDataStart,
- targetVector.getDataBuffer().memoryAddress() + targetDataSize,
- deltaDataSize);
+ targetVector
Review Comment:
Also, only the data copy is checked now. The validity copy via `concatBits`,
the offset buffer `MemoryUtil.copyMemory`, and the offset-rebasing loop still
read the delta without bounds checks. A delta whose offset buffer is shorter
than `(valueCount + 1) * 4` bytes, or whose validity buffer is shorter than
`valueCount` bits, would still read out-of-bounds native memory.
Is it worth validating those buffer sizes up front too?
--
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]
