From ea44821935a8945780f6e924115357ed7dbc03ae Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Wed, 12 Aug 2026 11:30:10 +0000
Subject: [PATCH] [#860] Serve JDBC cursor repositioning from the fetched buffer and grow batches adaptively (#863)
---
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java | 132 +++++++++++++++++++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java | 63 ++++++++++++++-
2 files changed, 189 insertions(+), 6 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
index b8cd126..fcf3737 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
@@ -321,12 +321,26 @@
}catch (SQLException e) {
throw new StorageRuntimeException(e);
}
+ }else if (driverName.contains("oracle")) {
+ try {
+ // oracle has no "create index if not exists"; unquoted identifiers are stored in uppercase
+ if (!isExistsIndex(tableName.toUpperCase(),"k_"+tableName.substring("opendj_".length()))) {
+ try (final PreparedStatement statement=con.prepareStatement("create index k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){
+ execute(statement);
+ con.commit();
+ }
+ }
+ }catch (SQLException e) {
+ throw new StorageRuntimeException(e);
+ }
}
+ // mssql: k is varbinary(max), which cannot be an index key column - cursor batches stay unindexed there
}
}
boolean isExistsIndex(String tableName, String indexName) throws SQLException {
- try (final ResultSet rs = con.getMetaData().getIndexInfo(null, null, tableName, false, false)) {
+ // approximate=true: with false the oracle driver runs ANALYZE on every call
+ try (final ResultSet rs = con.getMetaData().getIndexInfo(null, null, tableName, false, true)) {
while (rs.next()) {
if (indexName.equalsIgnoreCase(rs.getString("INDEX_NAME"))) {
return true;
@@ -449,13 +463,23 @@
}
}
- // Iterates in batches of "fetchsize" records via keyset pagination ("where k>? order by k limit n"):
+ static int compareKeys(byte[] key1, byte[] key2) {
+ return ByteString.wrap(key1).compareTo(key2, 0, key2.length);
+ }
+
+ // Iterates in batches via keyset pagination ("where k>? order by k limit n"):
// scrollable ResultSet is not an option, the postgres/mysql drivers materialize it entirely in memory.
- private final class CursorImpl implements Cursor<ByteString, ByteString> {
+ // Batches start at "fetchsize.initial" and grow geometrically to "fetchsize" while the reads stay
+ // sequential: most cursors read only a few rows, and eagerly fetching the maximum made every
+ // repositioning transfer "fetchsize" rows over the network (#860).
+ final class CursorImpl implements Cursor<ByteString, ByteString> {
final Connection con;
final String tableName;
final boolean isReadOnly;
final int batchSize=Math.max(1,Integer.getInteger("org.openidentityplatform.opendj.jdbc.fetchsize",1000));
+ final int initialBatchSize=Math.min(batchSize,Math.max(1,Integer.getInteger("org.openidentityplatform.opendj.jdbc.fetchsize.initial",32)));
+ int nextBatchSize=initialBatchSize;
+ long fetchCount;
final String limitClause;
final ArrayDeque<byte[][]> buffer=new ArrayDeque<>();
@@ -472,7 +496,14 @@
? " limit ?,?" : " offset ? rows fetch next ? rows only";
}
+ int adaptiveBatchSize() {
+ final int size=nextBatchSize;
+ nextBatchSize=Math.min(batchSize,size*4);
+ return size;
+ }
+
boolean fetchBatch(String condition, byte[] dbKey, long offset, boolean descending, int limit) {
+ fetchCount++;
buffer.clear();
try (final PreparedStatement statement=con.prepareStatement("select k,v from "+tableName
+(condition!=null?" where k"+condition+"?":"")
@@ -504,7 +535,7 @@
@Override
public boolean next() {
- if (buffer.isEmpty() && !fetchBatch(currentKeyDb==null?null:">",currentKeyDb,0,false,batchSize)) {
+ if (buffer.isEmpty() && !fetchBatch(currentKeyDb==null?null:">",currentKeyDb,0,false,adaptiveBatchSize())) {
defined=false;
return false;
}
@@ -558,7 +589,23 @@
@Override
public boolean positionToKeyOrNext(ByteSequence key) {
- if (fetchBatch(">=",real2db(key.toByteArray()),0,false,batchSize)) {
+ final byte[] target=real2db(key.toByteArray());
+ // Forward repositioning within the already-fetched range is served from the buffer: buffered
+ // rows are the contiguous sorted rows following the current one (byte order matches the
+ // database binary collation), so the first row >= target is guaranteed to be among them.
+ if (!buffer.isEmpty() && currentKeyDb!=null
+ && compareKeys(target,currentKeyDb)>0
+ && compareKeys(target,buffer.peekLast()[0])<=0) {
+ while (compareKeys(buffer.peek()[0],target)<0) {
+ buffer.poll();
+ }
+ advanceFromBuffer();
+ return true;
+ }
+ if (!buffer.isEmpty()) { // jumped outside the buffered range: random access, back to small batches
+ nextBatchSize=initialBatchSize;
+ }
+ if (fetchBatch(">=",target,0,false,adaptiveBatchSize())) {
advanceFromBuffer();
return true;
}
@@ -575,6 +622,7 @@
try(final ResultSet rc=executeResultSet(statement)) {
if (rc.next()) {
buffer.clear();
+ nextBatchSize=initialBatchSize;
currentKeyDb=real2db(real);
currentKey=ByteString.wrap(real);
currentValue=ByteString.wrap(rc.getBytes("v"));
@@ -601,7 +649,10 @@
@Override
public boolean positionToIndex(int index) {
- if (index>=0 && fetchBatch(null,null,index,false,batchSize)) {
+ if (!buffer.isEmpty()) { // absolute jump: random access, back to small batches
+ nextBatchSize=initialBatchSize;
+ }
+ if (index>=0 && fetchBatch(null,null,index,false,adaptiveBatchSize())) {
advanceFromBuffer();
return true;
}
diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
index e4463b3..472f62f 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
@@ -16,6 +16,7 @@
package org.opends.server.backends.jdbc;
import org.forgerock.opendj.ldap.ByteString;
+import org.forgerock.opendj.ldap.ByteStringBuilder;
import org.forgerock.opendj.server.config.server.JDBCBackendCfg;
import org.opends.server.backends.pluggable.PluggableBackendImplTestCase;
import org.opends.server.backends.pluggable.spi.AccessMode;
@@ -137,6 +138,137 @@
return ByteString.valueOfUtf8("value" + i);
}
+ /**
+ * Forward repositioning inside the already-fetched batch must be served from the buffer without SQL,
+ * and batch sizes must grow from "fetchsize.initial" to "fetchsize" on sequential reads (#860).
+ */
+ @Test
+ public void testPositionToKeyOrNextServedFromBuffer() throws Exception {
+ System.setProperty("org.openidentityplatform.opendj.jdbc.fetchsize", "8");
+ System.setProperty("org.openidentityplatform.opendj.jdbc.fetchsize.initial", "2");
+ final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
+ final TreeName tree = new TreeName("testCursorBuffer", "tree");
+ try {
+ storage.open(AccessMode.READ_WRITE);
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ txn.openTree(tree, true);
+ for (int i = 0; i < 40; i++) {
+ txn.put(tree, key(i), value(i));
+ }
+ }
+ });
+ storage.read(new ReadOperation<Void>() {
+ @Override
+ public Void run(ReadableTransaction txn) throws Exception {
+ try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
+ final JDBCStorage.CursorImpl impl = (JDBCStorage.CursorImpl) cursor;
+
+ assertTrue(cursor.next()); // fetch #1: initial batch of 2 (key00, key01)
+ assertEquals(cursor.getKey(), key(0));
+ assertEquals(impl.fetchCount, 1);
+ assertTrue(cursor.next()); // key01 is buffered
+ assertEquals(impl.fetchCount, 1);
+ assertTrue(cursor.next()); // fetch #2: grown batch of 8 (key02..key09)
+ assertEquals(cursor.getKey(), key(2));
+ assertEquals(impl.fetchCount, 2);
+
+ // forward repositioning within the fetched range must not run SQL
+ assertTrue(cursor.positionToKeyOrNext(key(5)));
+ assertEquals(cursor.getKey(), key(5));
+ assertEquals(cursor.getValue(), value(5));
+ assertEquals(impl.fetchCount, 2);
+ assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key051"))); // between rows
+ assertEquals(cursor.getKey(), key(6));
+ assertEquals(impl.fetchCount, 2);
+ assertTrue(cursor.positionToKeyOrNext(key(9))); // last buffered row
+ assertEquals(cursor.getKey(), key(9));
+ assertEquals(impl.fetchCount, 2);
+
+ assertTrue(cursor.positionToKeyOrNext(key(20))); // fetch #3: beyond the buffer
+ assertEquals(cursor.getKey(), key(20));
+ assertEquals(impl.fetchCount, 3);
+ assertTrue(cursor.positionToKeyOrNext(key(1))); // fetch #4: backward
+ assertEquals(cursor.getKey(), key(1));
+ assertEquals(impl.fetchCount, 4);
+
+ // emulate DN2ID.ChildrenCursor: reposition to currentKey+0x01 for every row.
+ // Before the fix every reposition re-fetched a full batch: 38 fetches here.
+ final long fetchesBefore = impl.fetchCount;
+ int rows = 1; // standing on key01
+ while (cursor.positionToKeyOrNext(
+ new ByteStringBuilder().appendBytes(cursor.getKey()).appendByte(0x01).toByteString())) {
+ rows++;
+ }
+ assertEquals(rows, 39); // key01..key39
+ assertTrue(impl.fetchCount - fetchesBefore <= 8,
+ "sibling scan took " + (impl.fetchCount - fetchesBefore) + " fetches");
+ }
+ return null;
+ }
+ });
+ } finally {
+ System.clearProperty("org.openidentityplatform.opendj.jdbc.fetchsize");
+ System.clearProperty("org.openidentityplatform.opendj.jdbc.fetchsize.initial");
+ try {
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ txn.deleteTree(tree);
+ }
+ });
+ } catch (Exception ignored) {}
+ storage.close();
+ }
+ }
+
+ /** Buffer-served repositioning relies on the database collating keys in unsigned byte order. */
+ @Test
+ public void testCursorKeyOrderIsUnsigned() throws Exception {
+ final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
+ final TreeName tree = new TreeName("testCursorOrder", "tree");
+ final ByteString low = ByteString.valueOfBytes(new byte[] { 0x7F });
+ final ByteString high = ByteString.valueOfBytes(new byte[] { (byte) 0x80, 0x01 });
+ try {
+ storage.open(AccessMode.READ_WRITE);
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ txn.openTree(tree, true);
+ txn.put(tree, low, value(1));
+ txn.put(tree, high, value(2));
+ }
+ });
+ storage.read(new ReadOperation<Void>() {
+ @Override
+ public Void run(ReadableTransaction txn) throws Exception {
+ try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
+ // with a signed collation 0x80 would sort before 0x7F and these would fail
+ assertTrue(cursor.next());
+ assertEquals(cursor.getKey(), low);
+ assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfBytes(new byte[] { (byte) 0x80 })));
+ assertEquals(cursor.getKey(), high);
+ assertFalse(cursor.next());
+ assertTrue(cursor.positionToLastKey());
+ assertEquals(cursor.getKey(), high);
+ }
+ return null;
+ }
+ });
+ } finally {
+ try {
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ txn.deleteTree(tree);
+ }
+ });
+ } catch (Exception ignored) {}
+ storage.close();
+ }
+ }
+
/** Cursor operations must keep working when the tree spans several "fetchsize" batches. */
@Test
public void testCursorCrossesFetchSizeBatches() throws Exception {
--
Gitblit v1.10.0