fix binaryconsume consuming null value

Author: hnwyllmm

adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/BinaryConsumer.java

@@ -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)) {
@@ -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() {
@@ -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();
     }
   }

adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/BinaryConsumerTest.java

@@ -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;
@@ -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();
           }
         },
@@ -119,5 +124,14 @@ 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);
   }
 }