Arawoof06 commented on code in PR #1364:
URL: https://github.com/apache/arrow-java/pull/1364#discussion_r4304112151
##########
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:
Agreed. Added a `validateDelta` pass at the top of both visits that runs
before any realloc or `concatBits`, so a rejected delta now leaves the target
as it was. The test checks that too (see below).
##########
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:
Yes, worth it. `validateDelta` now checks the validity buffer against
`getValidityBufferSizeFromCount(valueCount)` and the offset buffer against
`(valueCount + 1) * OFFSET_WIDTH` before anything reads them, so `concatBits`,
the offset `copyMemory` and the rebasing loop only run on buffers that cover
the value count.
##########
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:
Done. It walks the offsets once and requires `0 <= first`, non-decreasing
offsets throughout, and last `<=` data buffer capacity, so intermediate offsets
are covered too. `targetDataSize + deltaDataSize` uses `Math.addExact` now.
##########
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:
Same changes on the large path, with long offsets and `Math.addExact(long,
long)`.
##########
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:
The new checks are explicit capacity/offset comparisons rather than relying
on `ArrowBuf` bounds checking, so they apply with
`arrow.enable_unsafe_memory_access=true` as well, and I left the test
unconditional instead of adding an `assumeTrue`. After the failure it now
asserts the dictionary still has 2 values, `foo` and `bar`, and that validity
bits 2 and 3 are unset. With only the earlier `setBytes` change, that validity
assertion fails.
##########
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:
Added. The malformed case is parameterized over VarChar and LargeVarChar and
over four offset shapes: past the data buffer, decreasing intermediate offset,
last before first, and negative start. There is also a positive test for both
types with offsets `{3, 6, 9}` over `xxxbazqux`, which expects `foo, bar, baz,
qux`.
##########
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:
Fixed. The stream setup moved into a helper that closes everything in a
`finally`, and each test only holds the reader in try-with-resources.
--
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]
