GH-1478: Don't split surrogate pairs between ClobConsumer chunks

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

@@ -57,6 +57,24 @@ public void resetValueVector(VarCharVector vector) {
     this.currentIndex = 0;
   }
 
+  /**
+   * Reads at most {@code size} characters of the clob from {@code pos}. A chunk never ends with
+   * the high surrogate of a pair: it is left for the next chunk, so that the pair is encoded to
+   * UTF-8 as one character instead of two '?'.
+   */
+  private static String readChunk(Clob clob, long pos, int size, long length)
+      throws SQLException {
+    String chunk = clob.getSubString(pos, size);
+    if (chunk.isEmpty()) {
+      throw new SQLException("No characters read from clob at position " + pos + " of " + length);
+    }
+    int last = chunk.length() - 1;
+    if (last > 0 && pos + last < length && Character.isHighSurrogate(chunk.charAt(last))) {
+      return chunk.substring(0, last);
+    }
+    return chunk;
+  }
+
   /** Nullable consumer for clob data. */
   static class NullableClobConsumer extends ClobConsumer {
 
@@ -80,7 +98,7 @@ public void consume(ResultSet resultSet) throws SQLException {
           ArrowBuf offsetBuffer = vector.getOffsetBuffer();
           int startIndex = offsetBuffer.getInt(currentIndex * 4L);
           while (read <= length) {
-            String str = clob.getSubString(read, readSize);
+            String str = readChunk(clob, read, readSize, length);
             byte[] bytes = str.getBytes(StandardCharsets.UTF_8);
 
             while (dataBuffer.capacity() < startIndex + totalBytes + bytes.length) {
@@ -91,7 +109,7 @@ public void consume(ResultSet resultSet) throws SQLException {
                 bytes, 0, dataBuffer.memoryAddress() + startIndex + totalBytes, bytes.length);
 
             totalBytes += bytes.length;
-            read += readSize;
+            read += str.length();
           }
           offsetBuffer.setInt((currentIndex + 1) * 4L, startIndex + totalBytes);
           BitVectorHelper.setBit(vector.getValidityBuffer(), currentIndex);
@@ -124,7 +142,7 @@ public void consume(ResultSet resultSet) throws SQLException {
         ArrowBuf offsetBuffer = vector.getOffsetBuffer();
         int startIndex = offsetBuffer.getInt(currentIndex * 4L);
         while (read <= length) {
-          String str = clob.getSubString(read, readSize);
+          String str = readChunk(clob, read, readSize, length);
           byte[] bytes = str.getBytes(StandardCharsets.UTF_8);
 
           while (dataBuffer.capacity() < startIndex + totalBytes + bytes.length) {
@@ -135,7 +153,7 @@ public void consume(ResultSet resultSet) throws SQLException {
               bytes, 0, dataBuffer.memoryAddress() + startIndex + totalBytes, bytes.length);
 
           totalBytes += bytes.length;
-          read += readSize;
+          read += str.length();
         }
         offsetBuffer.setInt((currentIndex + 1) * 4L, startIndex + totalBytes);
         BitVectorHelper.setBit(vector.getValidityBuffer(), currentIndex);

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

@@ -56,6 +56,17 @@ public void testConsumeClobLargerThanDataBuffer() throws SQLException, IOExcepti
     assertConsume(false, repeat('a', INITIAL_VALUE_ALLOCATION * DEFAULT_RECORD_BYTE_COUNT * 4));
   }
 
+  @Test
+  public void testConsumeSurrogatePairsAcrossChunks() throws SQLException, IOException {
+    // The clob is read in chunks of 256 characters, so the first chunk of these values ends with
+    // the high surrogate of an emoji and the next one starts with its low surrogate.
+    String emoji = Character.toString(0x1F600);
+    for (String value : new String[] {repeat('a', 255) + emoji, "a" + emoji.repeat(300)}) {
+      assertConsume(true, value);
+      assertConsume(false, value);
+    }
+  }
+
   private static String repeat(char c, int count) {
     char[] chars = new char[count];
     java.util.Arrays.fill(chars, c);