From 2175f46934c41b30d5a26d9cc33d1f25467506db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?JB=20Onofr=C3=A9?= Date: Sun, 15 Nov 2026 18:31:36 +0100 Subject: [PATCH] GH-1478: Don't split surrogate pairs between ClobConsumer chunks Each chunk of a CLOB is encoded to UTF-8 on its own, so a character outside the BMP split between two chunks was encoded as "??". A chunk that ends with a high surrogate now leaves it for the next chunk, and the read position advances by the characters actually read. --- .../adapter/jdbc/consumer/ClobConsumer.java | 26 ++++++++++++++++--- .../jdbc/consumer/ClobConsumerTest.java | 11 ++++++++ 2 files changed, 33 insertions(+), 4 deletions(-) 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..1eb0fdd5bc 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 @@ -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); 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..400881b826 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 @@ -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);