diff --git a/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumer.java b/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumer.java index 7548f21608..54e7c0e9f9 100644 --- a/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumer.java +++ b/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; diff --git a/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumerTest.java b/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumerTest.java index eb1081dbc4..b6a7c58e58 100644 --- a/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumerTest.java +++ b/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;