Skip to content

GH-729: [JDBC] Fix BinaryConsumer consuming null value#730

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lidavidm merged 2 commits into from
Apr 28, 2025
Merged

GH-729: [JDBC] Fix BinaryConsumer consuming null value #730

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
662917a
fix binaryconsume consuming null value
hnwyllmm Apr 27, 2025
1718bc7
fix code format
hnwyllmm Apr 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 13 additions & 12 deletions adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/BinaryConsumer.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -51,13 +51,15 @@ public BinaryConsumer(VarBinaryVector vector, int index) {

/** consume a InputStream. */
public void consume(InputStream is) throws IOException {
while (currentIndex >= vector.getValueCapacity()) {
vector.reallocValidityAndOffsetBuffers();
}

final int startOffset = vector.getStartOffset(currentIndex);
final ArrowBuf offsetBuffer = vector.getOffsetBuffer();
int dataLength = 0;

if (is != null) {
while (currentIndex >= vector.getValueCapacity()) {
vector.reallocValidityAndOffsetBuffers();
}
final int startOffset = vector.getStartOffset(currentIndex);
final ArrowBuf offsetBuffer = vector.getOffsetBuffer();
int dataLength = 0;
int read;
while ((read = is.read(reuseBytes)) != -1) {
while (vector.getDataBuffer().capacity() < (startOffset + dataLength + read)) {
Expand All@@ -66,11 +68,12 @@ public void consume(InputStream is) throws IOException {
vector.getDataBuffer().setBytes(startOffset + dataLength, reuseBytes, 0, read);
dataLength += read;
}
offsetBuffer.setInt(
(currentIndex + 1) * ((long) VarBinaryVector.OFFSET_WIDTH), startOffset + dataLength);

BitVectorHelper.setBit(vector.getValidityBuffer(), currentIndex);
vector.setLastSet(currentIndex);
}
offsetBuffer.setInt(
(currentIndex + 1) * ((long) VarBinaryVector.OFFSET_WIDTH), startOffset + dataLength);
vector.setLastSet(currentIndex);
}

public void moveWriterPosition() {
Expand All@@ -95,9 +98,7 @@ public NullableBinaryConsumer(VarBinaryVector vector, int index) {
@Override
public void consume(ResultSet resultSet) throws SQLException, IOException {
InputStream is = resultSet.getBinaryStream(columnIndexInResultSet);
if (!resultSet.wasNull()) {
consume(is);
}
consume(is);
moveWriterPosition();
}
}
Expand Down
11 changes: 10 additions & 1 deletion adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/BinaryConsumerTest.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,6 +22,7 @@

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import org.apache.arrow.vector.BaseValueVector;
import org.apache.arrow.vector.VarBinaryVector;
import org.junit.jupiter.api.Test;
Expand DownExpand Up@@ -65,7 +66,11 @@ public void testConsumeInputStream(byte[][] values, boolean nullable) throws IOE
nullable,
binaryConsumer -> {
for (byte[] value : values) {
binaryConsumer.consume(new ByteArrayInputStream(value));
if (value != null) {
binaryConsumer.consume(new ByteArrayInputStream(value));
} else {
binaryConsumer.consume((InputStream) null);
}
binaryConsumer.moveWriterPosition();
}
},
Expand DownExpand Up@@ -119,5 +124,9 @@ public void testConsumeInputStream() throws IOException {
testRecords[i] = createBytes(DEFAULT_RECORD_BYTE_COUNT);
}
testConsumeInputStream(testRecords, false);

byte[] bytes1 = new byte[] {1, 2, 3};
byte[] bytes2 = new byte[] {4, 5, 6};
testConsumeInputStream(new byte[][] {bytes1, null, bytes2}, true);
}
}