From 49745f7a2c4b83688a87c7afef88c26c1eadf4e9 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 12 Aug 2026 09:55:56 +0300 Subject: [PATCH] [#860] Serve JDBC cursor repositioning from the fetched buffer and grow batches adaptively Every positionToKeyOrNext() discarded the buffer and re-fetched a full "fetchsize" batch (1000 rows), so DN2ID.ChildrenCursor transferred N x 1000 rows to enumerate N children. Forward repositioning within the already-fetched range is now served from the buffer without SQL, and batches start at "fetchsize.initial" (32) growing geometrically to "fetchsize" on sequential reads. Also create the cursor index on (k) for Oracle and stop the Oracle driver running ANALYZE in metadata checks. --- .../server/backends/jdbc/JDBCStorage.java | 63 ++++++++- .../opends/server/backends/jdbc/TestCase.java | 132 ++++++++++++++++++ 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 b8cd126165..fcf3737df4 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 @@ 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; @@ -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 { + // 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 { 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 buffer=new ArrayDeque<>(); @@ -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+"?":"") @@ -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; } @@ -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; } @@ -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")); @@ -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; } 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 e4463b356b..472f62fc9a 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 @@ 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() { + @Override + public Void run(ReadableTransaction txn) throws Exception { + try (final Cursor 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() { + @Override + public Void run(ReadableTransaction txn) throws Exception { + try (final Cursor 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 {