Skip to content

[#860] Serve JDBC cursor repositioning from the fetched buffer and grow batches adaptively #863

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vharseko merged 1 commit into from
Aug 12, 2026
+189 −6
Merged

[#860] Serve JDBC cursor repositioning from the fetched buffer and grow batches adaptively #863

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 57 additions & 6 deletions opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -321,12 +321,26 @@ public void openTree(TreeName treeName, boolean createOnDemand) {
}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;
Expand Down Expand Up @@ -449,13 +463,23 @@ public boolean delete(TreeName treeName, ByteSequence key) {
}
}

// 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<>();
Expand All @@ -472,7 +496,14 @@ public CursorImpl(boolean isReadOnly, Connection con, TreeName treeName) {
? " 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+"?":"")
Expand Down Expand Up @@ -504,7 +535,7 @@ void advanceFromBuffer() {

@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;
}
Expand Down Expand Up @@ -558,7 +589,23 @@ public void close() {

@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;
}
Expand All @@ -575,6 +622,7 @@ public boolean positionToKey(ByteSequence key) {
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"));
Expand All @@ -601,7 +649,10 @@ public boolean positionToLastKey() {

@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;
}
Expand Down
132 changes: 132 additions & 0 deletions opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -137,6 +138,137 @@ private static ByteString value(int i) {
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 {
Expand Down
Loading