GH-1477: Fill the offsets of NULL rows in ClobConsumer

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

@@ -70,6 +70,8 @@ public void consume(ResultSet resultSet) throws SQLException {
       Clob clob = resultSet.getClob(columnIndexInResultSet);
       if (!resultSet.wasNull()) {
         if (clob != null) {
+          // Fill the offsets of the NULL rows before this one, so it starts after the last value.
+          vector.fillEmpties(currentIndex);
           long length = clob.length();
 
           int read = 1;
@@ -114,6 +116,8 @@ public NonNullableClobConsumer(VarCharVector vector, int index) {
     public void consume(ResultSet resultSet) throws SQLException {
       Clob clob = resultSet.getClob(columnIndexInResultSet);
       if (clob != null) {
+        // Fill the offsets of the NULL rows before this one, so it starts after the last value.
+        vector.fillEmpties(currentIndex);
         long length = clob.length();
 
         int read = 1;

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

@@ -18,6 +18,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -28,6 +29,7 @@
 import org.apache.arrow.adapter.jdbc.ResultSetUtility;
 import org.apache.arrow.vector.BaseValueVector;
 import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.util.ValueVectorUtility;
 import org.junit.jupiter.api.Test;
 
 public class ClobConsumerTest extends AbstractConsumerTest {
@@ -44,11 +46,52 @@ private void assertConsume(boolean nullable, String value) throws SQLException,
     }
   }
 
+  private void assertConsumeRows(boolean nullable, String... values)
+      throws SQLException, IOException {
+    try (final VarCharVector vector = new VarCharVector("clob", allocator)) {
+      ClobConsumer consumer = ClobConsumer.createConsumer(vector, 0, nullable);
+      ClobRowResultSet resultSet = new ClobRowResultSet();
+      for (String value : values) {
+        resultSet.value = value;
+        consumer.consume(resultSet);
+      }
+      vector.setValueCount(values.length);
+      ValueVectorUtility.validateFull(vector);
+      for (int i = 0; i < values.length; i++) {
+        if (values[i] == null) {
+          assertTrue(vector.isNull(i), "row " + i);
+        } else {
+          assertArrayEquals(values[i].getBytes(StandardCharsets.UTF_8), vector.get(i), "row " + i);
+        }
+      }
+    }
+  }
+
   @Test
   public void testConsumeSmallClob() throws SQLException, IOException {
     assertConsume(false, repeat('a', DEFAULT_RECORD_BYTE_COUNT));
   }
 
+  @Test
+  public void testConsumeNullableClobsAfterNulls() throws SQLException, IOException {
+    // The values after a NULL row must be written after the previous value, not over it.
+    assertConsumeRows(true, null, "AAAA", null, null, "BB", null);
+  }
+
+  @Test
+  public void testConsumeNonNullableClobsAfterNullClobs() throws SQLException, IOException {
+    assertConsumeRows(false, null, "AAAA", null, null, "BB", null);
+  }
+
+  @Test
+  public void testConsumeMoreClobsThanValueCapacity() throws SQLException, IOException {
+    String[] values = new String[INITIAL_VALUE_ALLOCATION * 2];
+    for (int i = 0; i < values.length; i++) {
+      values[i] = i % 3 == 0 ? null : String.valueOf(i);
+    }
+    assertConsumeRows(true, values);
+  }
+
   @Test
   public void testConsumeClobLargerThanDataBuffer() throws SQLException, IOException {
     // A CLOB whose bytes exceed the VarCharVector's initial data-buffer capacity must grow the
@@ -81,6 +124,21 @@ public boolean wasNull() {
     }
   }
 
+  /** A {@link java.sql.ResultSet} whose CLOB is {@code value}, or NULL when {@code value} is null. */
+  private static class ClobRowResultSet extends ResultSetUtility.ThrowingResultSet {
+    private String value;
+
+    @Override
+    public Clob getClob(int columnIndex) {
+      return value == null ? null : new StringClob(value);
+    }
+
+    @Override
+    public boolean wasNull() {
+      return value == null;
+    }
+  }
+
   /** Minimal read-only {@link Clob} backed by a {@link String}. */
   private static class StringClob implements Clob {
     private final String value;