jbonofre commented on code in PR #1422:
URL: https://github.com/apache/arrow-java/pull/1422#discussion_r4463904226
##########
vector/src/main/java/org/apache/arrow/vector/ipc/message/MessageSerializer.java:
##########
@@ -411,50 +411,58 @@ public static ArrowRecordBatch deserializeRecordBatch(
*/
public static ArrowRecordBatch deserializeRecordBatch(RecordBatch
recordBatchFB, ArrowBuf body)
throws IOException {
- // Now read the body
- int nodesLength = recordBatchFB.nodesLength();
- List<ArrowFieldNode> nodes = new ArrayList<>();
- for (int i = 0; i < nodesLength; ++i) {
- FieldNode node = recordBatchFB.nodes(i);
- if ((int) node.length() != node.length() || (int) node.nullCount() !=
node.nullCount()) {
- throw new IOException(
- "Cannot currently deserialize record batches with "
- + "node length larger than INT_MAX records.");
+ try {
+ // Now read the body
+ int nodesLength = recordBatchFB.nodesLength();
+ List<ArrowFieldNode> nodes = new ArrayList<>();
+ for (int i = 0; i < nodesLength; ++i) {
+ FieldNode node = recordBatchFB.nodes(i);
+ if (node.length() < 0) {
+ throw new IOException("Record batch node " + i + " has negative
length " + node.length());
+ }
+ if (node.nullCount() < -1) {
+ throw new IOException(
+ "Record batch node " + i + " has invalid null count " +
node.nullCount());
+ }
+ if ((int) node.length() != node.length() || (int) node.nullCount() !=
node.nullCount()) {
+ throw new IOException(
+ "Cannot currently deserialize record batches with "
+ + "node length larger than INT_MAX records.");
+ }
+ nodes.add(new ArrowFieldNode(node.length(), node.nullCount()));
+ }
+ List<ArrowBuf> buffers = new ArrayList<>();
+ for (int i = 0; i < recordBatchFB.buffersLength(); ++i) {
+ Buffer bufferFB = recordBatchFB.buffers(i);
+ ArrowBuf vectorBuffer = body.slice(bufferFB.offset(),
bufferFB.length());
+ buffers.add(vectorBuffer);
}
- nodes.add(new ArrowFieldNode(node.length(), node.nullCount()));
- }
- List<ArrowBuf> buffers = new ArrayList<>();
- for (int i = 0; i < recordBatchFB.buffersLength(); ++i) {
- Buffer bufferFB = recordBatchFB.buffers(i);
- ArrowBuf vectorBuffer = body.slice(bufferFB.offset(), bufferFB.length());
- buffers.add(vectorBuffer);
- }
- ArrowBodyCompression bodyCompression =
- recordBatchFB.compression() == null
- ? NoCompressionCodec.DEFAULT_BODY_COMPRESSION
- : new ArrowBodyCompression(
- recordBatchFB.compression().codec(),
recordBatchFB.compression().method());
+ ArrowBodyCompression bodyCompression =
+ recordBatchFB.compression() == null
+ ? NoCompressionCodec.DEFAULT_BODY_COMPRESSION
+ : new ArrowBodyCompression(
+ recordBatchFB.compression().codec(),
recordBatchFB.compression().method());
- List<Long> variadicBufferCounts = new ArrayList<>();
- for (int i = 0; i < recordBatchFB.variadicBufferCountsLength(); i++) {
- variadicBufferCounts.add(recordBatchFB.variadicBufferCounts(i));
- }
+ List<Long> variadicBufferCounts = new ArrayList<>();
+ for (int i = 0; i < recordBatchFB.variadicBufferCountsLength(); i++) {
+ variadicBufferCounts.add(recordBatchFB.variadicBufferCounts(i));
+ }
- if ((int) recordBatchFB.length() != recordBatchFB.length()) {
- throw new IOException(
- "Cannot currently deserialize record batches with more than INT_MAX
records.");
+ if ((int) recordBatchFB.length() != recordBatchFB.length()) {
+ throw new IOException(
+ "Cannot currently deserialize record batches with more than
INT_MAX records.");
+ }
Review Comment:
The node checks above reject negative lengths, but a negative
`recordBatchFB.length()` still gets through: `(int) x != x` only catches values
above INT_MAX. A batch with `length = -100` reaches `VectorLoader.load`, which
calls `root.setRowCount(-100)` and then `setValueCount(-100)` on every vector,
where only an `assert` stands in the way.
Could we reject it here too?
```suggestion
if (recordBatchFB.length() < 0) {
throw new IOException("Record batch has negative length " +
recordBatchFB.length());
}
if ((int) recordBatchFB.length() != recordBatchFB.length()) {
throw new IOException(
"Cannot currently deserialize record batches with more than
INT_MAX records.");
}
```
##########
vector/src/main/java/org/apache/arrow/vector/ipc/message/MessageSerializer.java:
##########
@@ -411,50 +411,58 @@ public static ArrowRecordBatch deserializeRecordBatch(
*/
public static ArrowRecordBatch deserializeRecordBatch(RecordBatch
recordBatchFB, ArrowBuf body)
throws IOException {
- // Now read the body
- int nodesLength = recordBatchFB.nodesLength();
- List<ArrowFieldNode> nodes = new ArrayList<>();
- for (int i = 0; i < nodesLength; ++i) {
- FieldNode node = recordBatchFB.nodes(i);
- if ((int) node.length() != node.length() || (int) node.nullCount() !=
node.nullCount()) {
- throw new IOException(
- "Cannot currently deserialize record batches with "
- + "node length larger than INT_MAX records.");
+ try {
+ // Now read the body
+ int nodesLength = recordBatchFB.nodesLength();
+ List<ArrowFieldNode> nodes = new ArrayList<>();
+ for (int i = 0; i < nodesLength; ++i) {
+ FieldNode node = recordBatchFB.nodes(i);
+ if (node.length() < 0) {
+ throw new IOException("Record batch node " + i + " has negative
length " + node.length());
+ }
+ if (node.nullCount() < -1) {
+ throw new IOException(
+ "Record batch node " + i + " has invalid null count " +
node.nullCount());
+ }
+ if ((int) node.length() != node.length() || (int) node.nullCount() !=
node.nullCount()) {
+ throw new IOException(
+ "Cannot currently deserialize record batches with "
+ + "node length larger than INT_MAX records.");
+ }
+ nodes.add(new ArrowFieldNode(node.length(), node.nullCount()));
+ }
+ List<ArrowBuf> buffers = new ArrayList<>();
+ for (int i = 0; i < recordBatchFB.buffersLength(); ++i) {
+ Buffer bufferFB = recordBatchFB.buffers(i);
+ ArrowBuf vectorBuffer = body.slice(bufferFB.offset(),
bufferFB.length());
+ buffers.add(vectorBuffer);
}
- nodes.add(new ArrowFieldNode(node.length(), node.nullCount()));
- }
- List<ArrowBuf> buffers = new ArrayList<>();
- for (int i = 0; i < recordBatchFB.buffersLength(); ++i) {
- Buffer bufferFB = recordBatchFB.buffers(i);
- ArrowBuf vectorBuffer = body.slice(bufferFB.offset(), bufferFB.length());
- buffers.add(vectorBuffer);
- }
- ArrowBodyCompression bodyCompression =
- recordBatchFB.compression() == null
- ? NoCompressionCodec.DEFAULT_BODY_COMPRESSION
- : new ArrowBodyCompression(
- recordBatchFB.compression().codec(),
recordBatchFB.compression().method());
+ ArrowBodyCompression bodyCompression =
+ recordBatchFB.compression() == null
+ ? NoCompressionCodec.DEFAULT_BODY_COMPRESSION
+ : new ArrowBodyCompression(
+ recordBatchFB.compression().codec(),
recordBatchFB.compression().method());
- List<Long> variadicBufferCounts = new ArrayList<>();
- for (int i = 0; i < recordBatchFB.variadicBufferCountsLength(); i++) {
- variadicBufferCounts.add(recordBatchFB.variadicBufferCounts(i));
- }
+ List<Long> variadicBufferCounts = new ArrayList<>();
+ for (int i = 0; i < recordBatchFB.variadicBufferCountsLength(); i++) {
+ variadicBufferCounts.add(recordBatchFB.variadicBufferCounts(i));
+ }
- if ((int) recordBatchFB.length() != recordBatchFB.length()) {
- throw new IOException(
- "Cannot currently deserialize record batches with more than INT_MAX
records.");
+ if ((int) recordBatchFB.length() != recordBatchFB.length()) {
+ throw new IOException(
+ "Cannot currently deserialize record batches with more than
INT_MAX records.");
+ }
+ return new ArrowRecordBatch(
+ checkedCastToInt(recordBatchFB.length()),
+ nodes,
+ buffers,
+ bodyCompression,
+ variadicBufferCounts,
+ /*alignBuffers*/ true);
+ } finally {
+ body.getReferenceManager().release();
Review Comment:
`body` can be null here: `deserializeMessageBatch(MessageChannelReader)`
passes `result.getBodyBuffer()`, which stays null when the message has no body
(`MessageChannelReader#readNext`). This already threw NPE on `main`, but now
that it's in a `finally`, the NPE also replaces the new `IOException` for a
malformed batch, so callers get a NPE instead of the declared exception.
```suggestion
if (body != null) {
body.getReferenceManager().release();
}
```
--
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]
