Skip to content

[improve][ml] Optimize individual-ack filtering by skipping scans for batches without acked entries #26311

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

Open
void-ptr974 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[improve][ml] Optimize individual-ack filtering by skipping scans for batches without acked entries #26311

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f4d9668
[improve][ml] Optimize individual-ack filtering by skipping scans for…
void-ptr974 Aug 11, 2026
ff21455
[improve][ml] Harden individual-ack filtering boundaries
void-ptr974 Aug 15, 2026
5fa07af
[fix][test] Pin cross-ledger read filtering fallback
void-ptr974 Aug 22, 2026
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
30 changes: 23 additions & 7 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedCursorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2754,22 +2754,38 @@ private void updateLastMarkDeleteEntryToLatest(final Position newPosition,
* Given a list of entries, filter out the entries that have already been individually deleted.
*
* @param entries
* a list of entries
* a non-empty list of entries ordered by position; read paths normally return entries from one ledger
* @return a list of entries not containing deleted messages
*/
List<Entry> filterReadEntries(List<Entry> entries) {
lock.readLock().lock();
try {
Range<Position> entriesRange = Range.closed(entries.get(0).getPosition(),
entries.get(entries.size() - 1).getPosition());
Entry firstEntry = entries.get(0);
Entry lastEntry = entries.get(entries.size() - 1);
long firstLedgerId = firstEntry.getLedgerId();
long firstEntryId = firstEntry.getEntryId();
long lastLedgerId = lastEntry.getLedgerId();
long lastEntryId = lastEntry.getEntryId();
log.debug()
.attr("entriesRange", entriesRange)
.attr("firstLedgerId", firstLedgerId)
.attr("firstEntryId", firstEntryId)
.attr("lastLedgerId", lastLedgerId)
.attr("lastEntryId", lastEntryId)
.attr("deletedMessages", individualDeletedMessages)
.log("Filtering entries");
Range<Position> span = individualDeletedMessages.isEmpty() ? null : individualDeletedMessages.span();
if (span == null || !entriesRange.isConnected(span)) {
// Read batches are ordered and normally belong to one ledger. For an unexpected cross-ledger or
// descending batch, conservatively retain per-entry filtering.
boolean containsDeletedMessages = firstLedgerId != lastLedgerId
|| firstEntryId > lastEntryId
|| individualDeletedMessages.containsAny(firstLedgerId, firstEntryId, lastEntryId);
if (!containsDeletedMessages) {
// There are no individually deleted messages in this entry list, no need to perform filtering
log.debug().attr("entriesRange", entriesRange).log("No filtering needed for entries");
log.debug()
.attr("firstLedgerId", firstLedgerId)
.attr("firstEntryId", firstEntryId)
.attr("lastLedgerId", lastLedgerId)
.attr("lastEntryId", lastEntryId)
.log("No filtering needed for entries");
return entries;
} else {
// Remove from the entry list all the entries that were already marked for deletion
Expand Down
19 changes: 19 additions & 0 deletions managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/PositionRangeSet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,25 @@ public boolean contains(long ledgerId, long entryId) {
return false;
}

/**
* Returns {@code true} if any entry in the closed range {@code [lowerEntryId, upperEntryId]} of
* {@code ledgerId} is present in this set. Both bounds are inclusive (matching
* {@link #cardinality}), unlike the open lower bound of {@link #addOpenClosed}. A negative
* {@code lowerEntryId} is treated as {@code 0}. Returns {@code false} for an empty range or an
* absent ledger.
*/
boolean containsAny(long ledgerId, long lowerEntryId, long upperEntryId) {
Comment thread
lhotari marked this conversation as resolved.
if (rangeBitmapMap.isEmpty() || lowerEntryId > upperEntryId) {
return false;
}
LongBitmap bitmap = rangeBitmapMap.get(ledgerId);
if (bitmap == null) {
return false;
}
long nextPresentEntryId = bitmap.nextPresentValue(Math.max(0, lowerEntryId));
Comment thread
lhotari marked this conversation as resolved.
return nextPresentEntryId != -1 && nextPresentEntryId <= upperEntryId;
}

@Override
public Range<Position> rangeContaining(long ledgerId, long entryId) {
LongBitmap rangeBitmap = rangeBitmapMap.get(ledgerId);
Expand Down
128 changes: 128 additions & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedCursorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5487,6 +5487,134 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
ledger.close();
}

@Test
public void testFilterReadEntriesSkipsFilteringForGapBetweenIndividualAcks() throws Exception {
@Cleanup
ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
"testFilterReadEntriesSkipsFilteringForGapBetweenIndividualAcks");
@Cleanup
ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c");

List<Position> positions = new ArrayList<>();
for (int i = 0; i <= 100; i++) {
positions.add(ledger.addEntry(new byte[]{1}));
}
cursor.delete(Set.of(positions.get(1), positions.get(100)));

List<Entry> entries = new ArrayList<>();
for (int i = 40; i < 50; i++) {
Position position = positions.get(i);
Entry entry = mock(Entry.class);
when(entry.getPosition()).thenReturn(position);
when(entry.getLedgerId()).thenReturn(position.getLedgerId());
when(entry.getEntryId()).thenReturn(position.getEntryId());
entries.add(entry);
}

Range<Position> entriesRange = Range.closed(entries.get(0).getPosition(),
entries.get(entries.size() - 1).getPosition());
assertThat(entriesRange.isConnected(cursor.getIndividuallyDeletedMessagesSet().span())).isTrue();
for (Entry entry : entries) {
assertThat(cursor.getIndividuallyDeletedMessagesSet()
.contains(entry.getLedgerId(), entry.getEntryId())).isFalse();
}

List<Entry> filteredEntries = cursor.filterReadEntries(entries);

assertThat(filteredEntries).isSameAs(entries);
for (Entry entry : entries) {
verify(entry, never()).release();
}
}

@Test
public void testFilterReadEntriesFallsBackForCrossLedgerBatch() throws Exception {
@Cleanup
ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
"testFilterReadEntriesFallsBackForCrossLedgerBatch");
@Cleanup
ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c");

// Production read paths currently return ordered batches from a single ledger, so this cross-ledger batch
// is not expected in normal operation. If one is ever passed in, the conservative fallback must inspect
// every entry, remove and release the acknowledged entry, and retain the unacknowledged entries.
Position firstPosition = PositionFactory.create(1, 0);
Position deletedPosition = PositionFactory.create(2, 0);
Position lastPosition = PositionFactory.create(2, 1);
cursor.lock.writeLock().lock();
try {
cursor.getIndividuallyDeletedMessagesSet().addOpenClosed(
deletedPosition.getLedgerId(), deletedPosition.getEntryId() - 1,
deletedPosition.getLedgerId(), deletedPosition.getEntryId());
} finally {
cursor.lock.writeLock().unlock();
}

Entry firstEntry = mock(Entry.class);
when(firstEntry.getPosition()).thenReturn(firstPosition);
when(firstEntry.getLedgerId()).thenReturn(firstPosition.getLedgerId());
when(firstEntry.getEntryId()).thenReturn(firstPosition.getEntryId());
Entry deletedEntry = mock(Entry.class);
when(deletedEntry.getPosition()).thenReturn(deletedPosition);
when(deletedEntry.getLedgerId()).thenReturn(deletedPosition.getLedgerId());
when(deletedEntry.getEntryId()).thenReturn(deletedPosition.getEntryId());
Entry lastEntry = mock(Entry.class);
when(lastEntry.getPosition()).thenReturn(lastPosition);
when(lastEntry.getLedgerId()).thenReturn(lastPosition.getLedgerId());
when(lastEntry.getEntryId()).thenReturn(lastPosition.getEntryId());

List<Entry> filteredEntries = cursor.filterReadEntries(List.of(firstEntry, deletedEntry, lastEntry));

assertThat(filteredEntries).containsExactly(firstEntry, lastEntry);
verify(firstEntry, never()).release();
verify(deletedEntry).release();
verify(lastEntry, never()).release();
}

@Test
public void testFilterReadEntriesFallsBackForDescendingBatch() throws Exception {
@Cleanup
ManagedLedgerImpl ledger = (ManagedLedgerImpl) factory.open(
"testFilterReadEntriesFallsBackForDescendingBatch");
@Cleanup
ManagedCursorImpl cursor = (ManagedCursorImpl) ledger.openCursor("c");

// Production read paths currently return ordered batches, so this descending batch is not expected in
// normal operation. If one is ever passed in, the conservative fallback must inspect every entry, remove
// and release the acknowledged entry, and retain the unacknowledged entries.
Position firstPosition = PositionFactory.create(1, 10);
Position deletedPosition = PositionFactory.create(1, 5);
Position lastPosition = PositionFactory.create(1, 1);
cursor.lock.writeLock().lock();
try {
cursor.getIndividuallyDeletedMessagesSet().addOpenClosed(
deletedPosition.getLedgerId(), deletedPosition.getEntryId() - 1,
deletedPosition.getLedgerId(), deletedPosition.getEntryId());
} finally {
cursor.lock.writeLock().unlock();
}

Entry firstEntry = mock(Entry.class);
when(firstEntry.getPosition()).thenReturn(firstPosition);
when(firstEntry.getLedgerId()).thenReturn(firstPosition.getLedgerId());
when(firstEntry.getEntryId()).thenReturn(firstPosition.getEntryId());
Entry deletedEntry = mock(Entry.class);
when(deletedEntry.getPosition()).thenReturn(deletedPosition);
when(deletedEntry.getLedgerId()).thenReturn(deletedPosition.getLedgerId());
when(deletedEntry.getEntryId()).thenReturn(deletedPosition.getEntryId());
Entry lastEntry = mock(Entry.class);
when(lastEntry.getPosition()).thenReturn(lastPosition);
when(lastEntry.getLedgerId()).thenReturn(lastPosition.getLedgerId());
when(lastEntry.getEntryId()).thenReturn(lastPosition.getEntryId());

List<Entry> filteredEntries = cursor.filterReadEntries(List.of(firstEntry, deletedEntry, lastEntry));

assertThat(filteredEntries).containsExactly(firstEntry, lastEntry);
verify(firstEntry, never()).release();
verify(deletedEntry).release();
verify(lastEntry, never()).release();
}

@Test
public void testReadEntriesWithSkipDeletedEntries() throws Exception {
@Cleanup
Expand Down
109 changes: 109 additions & 0 deletions managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/PositionRangeSetTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.PositionFactory;
Expand Down Expand Up @@ -235,6 +236,114 @@ public void testSpanWithGuava() {
assertEquals(set.span(), Range.openClosed(pos(0, 97), pos(4, 20)));
}

@Test
public void testContainsAny() {
PositionRangeSet set = newSet();
set.addOpenClosed(1, 9, 1, 10);
set.addOpenClosed(1, 999, 1, 1000);
set.addOpenClosed(3, 19, 3, 20);

assertFalse(set.containsAny(1, 0, 9));
assertTrue(set.containsAny(1, 0, 10));
assertFalse(set.containsAny(1, 11, 999));
assertTrue(set.containsAny(1, 11, 1000));
assertFalse(set.containsAny(1, 1001, 2000));
assertFalse(set.containsAny(2, 0, 1000));
assertTrue(set.containsAny(3, 20, 20));
assertFalse(set.containsAny(3, 21, 20));
}

@Test
public void testContainsAnyBoundaries() {
PositionRangeSet set = newSet();
long maxEntryId = Integer.MAX_VALUE;

assertFalse(set.containsAny(1, 0, 0));

set.addOpenClosed(1, -1, 1, 0);
set.addOpenClosed(1, 9, 1, 10);
set.addOpenClosed(1, maxEntryId - 1, 1, maxEntryId);

assertTrue(set.containsAny(1, 0, 0));
Comment thread
lhotari marked this conversation as resolved.
assertTrue(set.containsAny(1, -1, 0));
assertFalse(set.containsAny(1, -5, -1));
assertFalse(set.containsAny(1, 1, 9));
assertTrue(set.containsAny(1, 9, 10));
assertTrue(set.containsAny(1, 10, 11));
assertFalse(set.containsAny(1, 11, maxEntryId - 1));
assertTrue(set.containsAny(1, maxEntryId - 1, maxEntryId));
assertTrue(set.containsAny(1, maxEntryId, maxEntryId));
assertFalse(set.containsAny(1, maxEntryId + 1, maxEntryId + 1));
}

@Test
public void testContainsAnyRandomizedAgainstBooleanOracle() {
int ledgerCount = 16;
int entriesPerLedger = 4096;
int seedsPerDensity = 32;
int queriesPerSeed = 500;
for (int density : new int[]{0, 1, 10, 50, 90, 100}) {
for (int seedIndex = 0; seedIndex < seedsPerDensity; seedIndex++) {
long seed = 0x5EEDL + density * 1_000_003L + seedIndex * 104_729L;
Random random = new Random(seed);
boolean[][] deletedEntries = new boolean[ledgerCount][entriesPerLedger];
PositionRangeSet set = newSet();

for (int ledgerId = 0; ledgerId < ledgerCount; ledgerId++) {
boolean ledgerHasDeletedEntries = random.nextInt(4) != 0;
for (int entryId = 0; entryId < entriesPerLedger; entryId++) {
deletedEntries[ledgerId][entryId] = ledgerHasDeletedEntries
&& random.nextInt(100) < density;
}
addDeletedRanges(set, deletedEntries[ledgerId], ledgerId);
}

for (int query = 0; query < queriesPerSeed; query++) {
int ledgerId = random.nextInt(10) == 0
? ledgerCount + random.nextInt(2) : random.nextInt(ledgerCount);
int firstEntryId = random.nextInt(entriesPerLedger);
int batchSize = 1 + random.nextInt(100);
int lastEntryId = Math.min(entriesPerLedger - 1, firstEntryId + batchSize - 1);

long expectedNextEntryId = nextDeletedEntryId(deletedEntries, ledgerId, firstEntryId);
boolean expected = expectedNextEntryId != -1 && expectedNextEntryId <= lastEntryId;
boolean actual = set.containsAny(ledgerId, firstEntryId, lastEntryId);
assertEquals(actual, expected,
"seed=" + seed + ", query=" + query + ", range="
+ ledgerId + ":" + firstEntryId + ".." + ledgerId + ":" + lastEntryId);
}
}
}
}

private static void addDeletedRanges(PositionRangeSet set, boolean[] deletedEntries, long ledgerId) {
int entryId = 0;
while (entryId < deletedEntries.length) {
while (entryId < deletedEntries.length && !deletedEntries[entryId]) {
entryId++;
}
int firstDeletedEntryId = entryId;
while (entryId < deletedEntries.length && deletedEntries[entryId]) {
entryId++;
}
if (firstDeletedEntryId < entryId) {
set.addOpenClosed(ledgerId, firstDeletedEntryId - 1L, ledgerId, entryId - 1L);
}
}
}

private static long nextDeletedEntryId(boolean[][] deletedEntries, int ledgerId, int firstEntryId) {
if (ledgerId < 0 || ledgerId >= deletedEntries.length) {
return -1;
}
for (int entryId = firstEntryId; entryId < deletedEntries[ledgerId].length; entryId++) {
if (deletedEntries[ledgerId][entryId]) {
return entryId;
}
}
return -1;
}

@Test
public void testFirstRange() {
PositionRangeSet set = newSet();
Expand Down
Loading
Loading