[#860] Serve JDBC cursor repositioning from the fetched buffer and grow bat...

Valery Kharseko

19 hours ago ea44821935a8945780f6e924115357ed7dbc03ae

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

 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;
            }

 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 {