/*
|
* The contents of this file are subject to the terms of the Common Development and
|
* Distribution License (the License). You may not use this file except in compliance with the
|
* License.
|
*
|
* You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
|
* specific language governing permission and limitations under the License.
|
*
|
* When distributing Covered Software, include this CDDL Header Notice in each file and include
|
* the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
|
* Header, with the fields enclosed by brackets [] replaced by your own identifying
|
* information: "Portions Copyright [year] [name of copyright owner]".
|
*
|
* Copyright 2023-2026 3A Systems, LLC.
|
*/
|
package org.opends.server.backends.cassandra;
|
|
import static org.mockito.Mockito.when;
|
import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
|
import static org.testng.Assert.assertEquals;
|
import static org.testng.Assert.assertFalse;
|
import static org.testng.Assert.assertTrue;
|
import static org.testng.Assert.fail;
|
|
import org.forgerock.opendj.ldap.ByteString;
|
import org.forgerock.opendj.ldap.ByteStringBuilder;
|
import org.forgerock.opendj.server.config.server.CASBackendCfg;
|
import org.opends.server.backends.pluggable.PluggableBackendImplTestCase;
|
import org.opends.server.backends.pluggable.spi.AccessMode;
|
import org.opends.server.backends.pluggable.spi.Cursor;
|
import org.opends.server.backends.pluggable.spi.ReadOperation;
|
import org.opends.server.backends.pluggable.spi.ReadableTransaction;
|
import org.opends.server.backends.pluggable.spi.TreeName;
|
import org.opends.server.backends.pluggable.spi.WriteOperation;
|
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
|
import org.testcontainers.DockerClientFactory;
|
import org.testcontainers.containers.CassandraContainer;
|
import org.testng.SkipException;
|
import org.testng.annotations.AfterClass;
|
import org.testng.annotations.Test;
|
|
import com.datastax.oss.driver.api.core.AllNodesFailedException;
|
import com.datastax.oss.driver.api.core.CqlSession;
|
import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
|
|
import java.net.InetSocketAddress;
|
import java.util.NoSuchElementException;
|
|
//docker run --rm -it -p 9042:9042 --name cassandra cassandra
|
|
//TestListener refuses a class that declares test methods of its own without sequential=true,
|
//and the cursor tests below share one storage, so they must not be interleaved either
|
@Test(groups = { "precommit", "pluggablebackend" }, sequential = true)
|
public class TestCase extends PluggableBackendImplTestCase<CASBackendCfg> {
|
|
private static final String PAGE_INITIAL = "org.openidentityplatform.opendj.cassandra.fetchsize.initial";
|
private static final String PAGE_MAX = "org.openidentityplatform.opendj.cassandra.fetchsize";
|
|
CassandraContainer cassandraContainer;
|
@Override
|
protected Backend createBackend() {
|
if(DockerClientFactory.instance().isDockerAvailable()) {
|
cassandraContainer = new CassandraContainer<>("cassandra:latest").withExposedPorts(9042);
|
cassandraContainer.start();
|
InetSocketAddress contactPoint = cassandraContainer.getContactPoint();
|
final String contactPointString = String.format("%s:%s", contactPoint.getHostName(), contactPoint.getPort());
|
System.setProperty("datastax-java-driver.basic.contact-points.0", contactPointString);
|
System.setProperty("datastax-java-driver.basic.load-balancing-policy.local-datacenter", cassandraContainer.getLocalDatacenter());
|
}
|
|
//test allow cassandra
|
try(CqlSession session=CqlSession.builder()
|
.withConfigLoader(DriverConfigLoader.fromDefaults(CASStorage.class.getClassLoader()))
|
.build()){
|
session.close();
|
}catch (AllNodesFailedException e) {
|
throw new SkipException("run before test: docker run --rm -it -p 9042:9042 --name cassandra cassandra");
|
}
|
return new Backend();
|
}
|
|
@Override
|
protected CASBackendCfg createBackendCfg() {
|
CASBackendCfg backendCfg = mockCfg(CASBackendCfg.class);
|
when(backendCfg.getBackendId()).thenReturn("CASTestCase");
|
when(backendCfg.getDBDirectory()).thenReturn("CASTestCase");
|
return backendCfg;
|
}
|
|
@AfterClass
|
@Override
|
public void cleanUp() throws Exception {
|
super.cleanUp();
|
if(cassandraContainer != null) {
|
cassandraContainer.close();
|
}
|
}
|
|
private static ByteString key(int i) {
|
return ByteString.valueOfUtf8(String.format("key%02d", i));
|
}
|
|
private static ByteString value(int i) {
|
return ByteString.valueOfUtf8("value" + i);
|
}
|
|
/**
|
* A cursor reads its page sizes when the storage is built, so pinning them here keeps the query
|
* counts below independent of the driver default (5000 rows) and of the storage defaults.
|
*/
|
private CASStorage openStorage(int initialPage, int maxPage) throws Exception {
|
System.setProperty(PAGE_INITIAL, String.valueOf(initialPage));
|
System.setProperty(PAGE_MAX, String.valueOf(maxPage));
|
try {
|
final CASStorage storage = new CASStorage(createBackendCfg(), null);
|
storage.open(AccessMode.READ_WRITE);
|
return storage;
|
} finally {
|
System.clearProperty(PAGE_INITIAL);
|
System.clearProperty(PAGE_MAX);
|
}
|
}
|
|
/** Rows left behind by an interrupted run would break the counts, so the tree starts empty. */
|
private static void fill(CASStorage storage, final TreeName tree, final int rows) throws Exception {
|
storage.write(new WriteOperation() {
|
@Override
|
public void run(WriteableTransaction txn) throws Exception {
|
txn.deleteTree(tree);
|
for (int i = 0; i < rows; i++) {
|
txn.put(tree, key(i), value(i));
|
}
|
}
|
});
|
}
|
|
private static void dropTree(CASStorage storage, final TreeName tree) {
|
try {
|
storage.write(new WriteOperation() {
|
@Override
|
public void run(WriteableTransaction txn) throws Exception {
|
txn.deleteTree(tree);
|
}
|
});
|
} catch (Exception e) { //a failed cleanup must be visible, but must not hide a test failure
|
System.err.println("cannot drop " + tree + ": " + e);
|
}
|
}
|
|
/**
|
* The driver ResultSet is consumed once and cannot be rewound, so every repositioning that is
|
* not a forward move within the already-fetched rows must run a new server-side slice query.
|
* The old implementation "restarted" the iterator via rc.iterator(), which is a no-op: backward
|
* repositioning returned the wrong row and positionToIndex counted from the current position.
|
*/
|
@Test
|
public void testCursorReposition() throws Exception {
|
final CASStorage storage = openStorage(32, 1000);
|
final TreeName tree = new TreeName("testCursorReposition", "tree");
|
try {
|
fill(storage, tree, 40);
|
storage.read(new ReadOperation<Void>() {
|
@Override
|
public Void run(ReadableTransaction txn) throws Exception {
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
final CASStorage.CursorImpl impl = (CASStorage.CursorImpl) cursor;
|
assertEquals(impl.queryCount, 0); // opening a cursor runs no query
|
|
assertTrue(cursor.positionToKeyOrNext(key(5))); // server-side seek
|
assertEquals(cursor.getKey(), key(5));
|
assertEquals(impl.queryCount, 1);
|
assertTrue(cursor.positionToKeyOrNext(key(5))); // same key: stays, no query
|
assertEquals(cursor.getKey(), key(5));
|
assertEquals(impl.queryCount, 1);
|
assertTrue(cursor.positionToKeyOrNext(key(9))); // forward: served from fetched rows
|
assertEquals(cursor.getKey(), key(9));
|
assertEquals(cursor.getValue(), value(9));
|
assertEquals(impl.queryCount, 1);
|
|
// backward: the old no-op "restart" returned the next remaining row instead
|
assertTrue(cursor.positionToKeyOrNext(key(2)));
|
assertEquals(cursor.getKey(), key(2));
|
assertEquals(cursor.getValue(), value(2));
|
assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key021"))); // between rows
|
assertEquals(cursor.getKey(), key(3));
|
|
assertTrue(cursor.positionToKey(key(1))); // backward exact match
|
assertEquals(cursor.getKey(), key(1));
|
final long queries = impl.queryCount;
|
assertFalse(cursor.positionToKey(ByteString.valueOfUtf8("key011"))); // missing key
|
assertFalse(cursor.isDefined());
|
assertEquals(impl.queryCount, queries); // the miss was decided within the page
|
assertTrue(cursor.next()); // a miss stops just before the next key (like pdb)
|
assertEquals(cursor.getKey(), key(2));
|
assertTrue(cursor.positionToKey(key(1)));
|
assertTrue(cursor.next()); // next() continues right after the positioned key (DN2ID)
|
assertEquals(cursor.getKey(), key(2));
|
|
// positionToIndex counts from the first row, not from the current position
|
assertTrue(cursor.positionToIndex(0));
|
assertEquals(cursor.getKey(), key(0));
|
assertTrue(cursor.positionToIndex(39));
|
assertEquals(cursor.getKey(), key(39));
|
assertFalse(cursor.positionToIndex(40));
|
|
assertTrue(cursor.positionToLastKey()); // LIMIT 1 query, no partition scan
|
assertEquals(cursor.getKey(), key(39));
|
assertFalse(cursor.next());
|
assertTrue(cursor.positionToKeyOrNext(key(0))); // reposition after exhaustion
|
assertEquals(cursor.getKey(), key(0));
|
assertFalse(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key99"))); // beyond last
|
|
// VLVIndex.evaluateVLVRequestByAssertion: seek to the assertion, then to the start
|
assertTrue(cursor.positionToKeyOrNext(key(20)) && cursor.positionToIndex(0));
|
assertEquals(cursor.getKey(), key(0));
|
}
|
|
// EntryContainer.deleteSubtree/renameSubtree walk an ascending key list on one
|
// shared id2entry cursor: forward moves must be served from the fetched rows,
|
// otherwise every entry costs a page-sized slice of full entries
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
final CASStorage.CursorImpl impl = (CASStorage.CursorImpl) cursor;
|
for (int i = 0; i < 40; i++) {
|
assertTrue(cursor.positionToKey(key(i)), "missing " + key(i));
|
assertEquals(cursor.getValue(), value(i));
|
}
|
assertEquals(impl.queryCount, 2, "ascending walk took " + impl.queryCount + " queries");
|
}
|
|
// DN2ID.ChildrenCursor: reposition to currentKey+0x01 for every row; forward
|
// repositioning is served from the fetched rows, so the scan stays at ~2 queries
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
final CASStorage.CursorImpl impl = (CASStorage.CursorImpl) cursor;
|
assertTrue(cursor.positionToKeyOrNext(key(0)));
|
int rows = 1;
|
while (cursor.positionToKeyOrNext(
|
new ByteStringBuilder().appendBytes(cursor.getKey()).appendByte(0x01).toByteString())) {
|
rows++;
|
}
|
assertEquals(rows, 40);
|
assertEquals(impl.queryCount, 3, "sibling scan took " + impl.queryCount + " queries");
|
}
|
return null;
|
}
|
});
|
} finally {
|
dropTree(storage, tree);
|
storage.close();
|
}
|
}
|
|
/**
|
* Everything a cursor does once its page runs out - continuing a scan, falling back to a slice,
|
* growing the page - only runs on trees bigger than one page, so the page is pinned small here.
|
* With the driver default of 5000 rows none of these branches would be exercised at all.
|
*/
|
@Test
|
public void testCursorPagingAcrossPages() throws Exception {
|
final CASStorage storage = openStorage(4, 8);
|
final TreeName tree = new TreeName("testCursorPaging", "tree");
|
try {
|
fill(storage, tree, 40);
|
storage.read(new ReadOperation<Void>() {
|
@Override
|
public Void run(ReadableTransaction txn) throws Exception {
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
final CASStorage.CursorImpl impl = (CASStorage.CursorImpl) cursor;
|
int rows = 0;
|
while (cursor.next()) { // the scan continues across page boundaries
|
assertEquals(cursor.getKey(), key(rows));
|
assertEquals(cursor.getValue(), value(rows));
|
rows++;
|
}
|
assertEquals(rows, 40);
|
assertFalse(cursor.next()); // and stops for good at the end of the partition
|
assertEquals(impl.pageSize, 8); // the page grew, but not past the maximum
|
assertTrue(impl.queryCount <= 8, "scan took " + impl.queryCount + " queries");
|
}
|
|
// forward repositioning falls back to a server-side slice when the page runs out
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
final CASStorage.CursorImpl impl = (CASStorage.CursorImpl) cursor;
|
assertTrue(cursor.positionToKeyOrNext(key(0)));
|
int rows = 1;
|
while (cursor.positionToKeyOrNext(
|
new ByteStringBuilder().appendBytes(cursor.getKey()).appendByte(0x01).toByteString())) {
|
assertEquals(cursor.getKey(), key(rows));
|
rows++;
|
}
|
assertEquals(rows, 40);
|
assertTrue(impl.queryCount <= 9, "sibling scan took " + impl.queryCount + " queries");
|
}
|
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
// rows far beyond the first page are still reachable in one seek
|
assertTrue(cursor.positionToKey(key(37)));
|
assertEquals(cursor.getValue(), value(37));
|
assertTrue(cursor.next());
|
assertEquals(cursor.getKey(), key(38));
|
assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfUtf8("key385")));
|
assertEquals(cursor.getKey(), key(39));
|
assertTrue(cursor.positionToIndex(39));
|
assertEquals(cursor.getKey(), key(39));
|
assertTrue(cursor.positionToIndex(20));
|
assertEquals(cursor.getKey(), key(20));
|
assertTrue(cursor.positionToLastKey());
|
assertEquals(cursor.getKey(), key(39));
|
assertFalse(cursor.next());
|
}
|
|
// DN2ID.openCursor0: position, then iterate with next() over several pages
|
try (final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree)) {
|
assertTrue(cursor.positionToKey(key(2)));
|
for (int i = 3; i < 40; i++) {
|
assertTrue(cursor.next());
|
assertEquals(cursor.getKey(), key(i));
|
}
|
assertFalse(cursor.next());
|
}
|
return null;
|
}
|
});
|
} finally {
|
dropTree(storage, tree);
|
storage.close();
|
}
|
}
|
|
/** Serving forward repositioning from fetched rows relies on the unsigned blob clustering order. */
|
@Test
|
public void testCursorKeyOrderIsUnsigned() throws Exception {
|
final CASStorage storage = openStorage(32, 1000);
|
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.write(new WriteOperation() {
|
@Override
|
public void run(WriteableTransaction txn) throws Exception {
|
txn.deleteTree(tree);
|
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)) {
|
final CASStorage.CursorImpl impl = (CASStorage.CursorImpl) cursor;
|
// with a signed collation 0x80 would sort before 0x7F and these would fail
|
assertTrue(cursor.next());
|
assertEquals(cursor.getKey(), low);
|
final long queries = impl.queryCount;
|
// {0x80} is above {0x7F} and below {0x80,0x01}, so this is a forward move served
|
// from the fetched rows: it is the client-side comparison that is checked here
|
assertTrue(cursor.positionToKeyOrNext(ByteString.valueOfBytes(new byte[] { (byte) 0x80 })));
|
assertEquals(cursor.getKey(), high);
|
assertEquals(impl.queryCount, queries);
|
assertFalse(cursor.next()); // {0x80,0x01} is the last key
|
assertTrue(cursor.positionToLastKey());
|
assertEquals(cursor.getKey(), high);
|
}
|
return null;
|
}
|
});
|
} finally {
|
dropTree(storage, tree);
|
storage.close();
|
}
|
}
|
|
/** A closed cursor is undefined and every navigation on it returns false, like EmptyCursor. */
|
@Test
|
public void testCursorAfterClose() throws Exception {
|
final CASStorage storage = openStorage(32, 1000);
|
final TreeName tree = new TreeName("testCursorClose", "tree");
|
try {
|
fill(storage, tree, 4);
|
storage.read(new ReadOperation<Void>() {
|
@Override
|
public Void run(ReadableTransaction txn) throws Exception {
|
final Cursor<ByteString, ByteString> cursor = txn.openCursor(tree);
|
assertTrue(cursor.positionToKeyOrNext(key(0)));
|
cursor.close();
|
assertFalse(cursor.isDefined());
|
assertFalse(cursor.next());
|
assertFalse(cursor.positionToKey(key(0)));
|
assertFalse(cursor.positionToKeyOrNext(key(0)));
|
assertFalse(cursor.positionToLastKey());
|
assertFalse(cursor.positionToIndex(0));
|
try {
|
cursor.getKey();
|
fail("a closed cursor has no key");
|
} catch (NoSuchElementException expected) {
|
// a closed cursor has no current row
|
}
|
cursor.close(); // closing twice is not an error
|
return null;
|
}
|
});
|
} finally {
|
dropTree(storage, tree);
|
storage.close();
|
}
|
}
|
}
|
