From 5a1bf1b3447fac39f1dad7b05859b33982fbe5ea Mon Sep 17 00:00:00 2001 From: abdul rawoof Date: Wed, 15 Jul 2026 14:35:14 +0530 Subject: [PATCH] GH-1230: fix out-of-bounds write in ClobConsumer buffer growth --- .../adapter/jdbc/consumer/ClobConsumer.java | 6 +- .../jdbc/consumer/ClobConsumerTest.java | 157 ++++++++++++++++++ 2 files changed, 161 insertions(+), 2 deletions(-) create mode 100644 adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumerTest.java 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 9fcdd42414..7548f21608 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 @@ -83,8 +83,9 @@ public void consume(ResultSet resultSet) throws SQLException { String str = clob.getSubString(read, readSize); byte[] bytes = str.getBytes(StandardCharsets.UTF_8); - while ((dataBuffer.writerIndex() + bytes.length) > dataBuffer.capacity()) { + while (dataBuffer.capacity() < startIndex + totalBytes + bytes.length) { vector.reallocDataBuffer(); + dataBuffer = vector.getDataBuffer(); } MemoryUtil.copyToMemory( bytes, 0, dataBuffer.memoryAddress() + startIndex + totalBytes, bytes.length); @@ -126,8 +127,9 @@ public void consume(ResultSet resultSet) throws SQLException { String str = clob.getSubString(read, readSize); byte[] bytes = str.getBytes(StandardCharsets.UTF_8); - while ((dataBuffer.writerIndex() + bytes.length) > dataBuffer.capacity()) { + while (dataBuffer.capacity() < startIndex + totalBytes + bytes.length) { vector.reallocDataBuffer(); + dataBuffer = vector.getDataBuffer(); } MemoryUtil.copyToMemory( bytes, 0, dataBuffer.memoryAddress() + startIndex + totalBytes, bytes.length); 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 new file mode 100644 index 0000000000..eb1081dbc4 --- /dev/null +++ b/adapter/jdbc/src/test/java/org/apache/arrow/adapter/jdbc/consumer/ClobConsumerTest.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.arrow.adapter.jdbc.consumer; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.IOException; +import java.io.InputStream; +import java.io.Reader; +import java.nio.charset.StandardCharsets; +import java.sql.Clob; +import java.sql.SQLException; +import org.apache.arrow.adapter.jdbc.ResultSetUtility; +import org.apache.arrow.vector.BaseValueVector; +import org.apache.arrow.vector.VarCharVector; +import org.junit.jupiter.api.Test; + +public class ClobConsumerTest extends AbstractConsumerTest { + + private static final int INITIAL_VALUE_ALLOCATION = BaseValueVector.INITIAL_VALUE_ALLOCATION; + private static final int DEFAULT_RECORD_BYTE_COUNT = 8; + + private void assertConsume(boolean nullable, String value) throws SQLException, IOException { + try (final VarCharVector vector = new VarCharVector("clob", allocator)) { + ClobConsumer consumer = ClobConsumer.createConsumer(vector, 0, nullable); + consumer.consume(new SingleClobResultSet(value)); + assertEquals(0, vector.getLastSet()); + assertArrayEquals(value.getBytes(StandardCharsets.UTF_8), vector.get(0)); + } + } + + @Test + public void testConsumeSmallClob() throws SQLException, IOException { + assertConsume(false, repeat('a', DEFAULT_RECORD_BYTE_COUNT)); + } + + @Test + public void testConsumeClobLargerThanDataBuffer() throws SQLException, IOException { + // A CLOB whose bytes exceed the VarCharVector's initial data-buffer capacity must grow the + // data buffer, not write past it. + assertConsume(false, repeat('a', INITIAL_VALUE_ALLOCATION * DEFAULT_RECORD_BYTE_COUNT * 4)); + } + + private static String repeat(char c, int count) { + char[] chars = new char[count]; + java.util.Arrays.fill(chars, c); + return new String(chars); + } + + /** A {@link java.sql.ResultSet} that returns a single non-null CLOB for any column. */ + private static class SingleClobResultSet extends ResultSetUtility.ThrowingResultSet { + private final Clob clob; + + SingleClobResultSet(String value) { + this.clob = new StringClob(value); + } + + @Override + public Clob getClob(int columnIndex) { + return clob; + } + + @Override + public boolean wasNull() { + return false; + } + } + + /** Minimal read-only {@link Clob} backed by a {@link String}. */ + private static class StringClob implements Clob { + private final String value; + + StringClob(String value) { + this.value = value; + } + + @Override + public long length() { + return value.length(); + } + + @Override + public String getSubString(long pos, int length) { + int start = (int) (pos - 1); + int end = Math.min(value.length(), start + length); + return value.substring(start, end); + } + + @Override + public Reader getCharacterStream() { + throw new UnsupportedOperationException(); + } + + @Override + public InputStream getAsciiStream() { + throw new UnsupportedOperationException(); + } + + @Override + public long position(String searchstr, long start) { + throw new UnsupportedOperationException(); + } + + @Override + public long position(Clob searchstr, long start) { + throw new UnsupportedOperationException(); + } + + @Override + public int setString(long pos, String str) { + throw new UnsupportedOperationException(); + } + + @Override + public int setString(long pos, String str, int offset, int len) { + throw new UnsupportedOperationException(); + } + + @Override + public java.io.OutputStream setAsciiStream(long pos) { + throw new UnsupportedOperationException(); + } + + @Override + public java.io.Writer setCharacterStream(long pos) { + throw new UnsupportedOperationException(); + } + + @Override + public void truncate(long len) { + throw new UnsupportedOperationException(); + } + + @Override + public void free() {} + + @Override + public Reader getCharacterStream(long pos, long length) { + throw new UnsupportedOperationException(); + } + } +}