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 {