Skip to content

perf: stream equality and range index scans instead of materializing every id #1298

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

Closed
brettwooldridge wants to merge 1 commit into from
Closed

perf: stream equality and range index scans instead of materializing every id #1298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
67 changes: 41 additions & 26 deletions nitrite/src/main/java/org/dizitart/no2/collection/operation/ReadOperations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,35 +111,43 @@ private void prepareLogicalFilter(LogicalFilter logicalFilter) {
}

private DocumentCursor createCursor(FindPlan findPlan) {
// -1 means "not an index scan"; the index branch records the exact id-set size here.
long[] indexedIdCount = { -1 };
RecordStream<Pair<NitriteId, Document>> recordStream = findSuitableStream(findPlan, indexedIdCount);
IndexScan scan = new IndexScan();
RecordStream<Pair<NitriteId, Document>> recordStream = findSuitableStream(findPlan, scan);
DocumentStream cursor = new DocumentStream(recordStream, processorChain);
cursor.setFindPlan(findPlan);
cursor.setCoveredCount(computeCoveredCount(findPlan, indexedIdCount[0]));
if (isCountCovered(findPlan)) {
if (findPlan.getIndexDescriptor() == null) {
// pure full scan over the whole collection
cursor.setCoveredCount(nitriteMap.size());
} else if (scan.lazyStream != null) {
// the ids are read lazily: count them from the index only if size() is asked
cursor.setCoveredCountSupplier(scan.lazyStream::countIds);
} else if (scan.idCount >= 0) {
// the index supplied the exact matching id set
cursor.setCoveredCount(scan.idCount);
}
}
return cursor;
}

/** What an index scan handed back: an exact id count, or the lazy stream, or neither. */
private static final class IndexScan {
private long idCount = -1;
private IndexedStream lazyStream;
}

/**
* Returns the exact match count when the query is fully answered without fetching documents,
* or {@code null} when the cursor must be drained to count. The count is exact only when
* nothing downstream drops or changes cardinality (a post-filter, skip, or limit); sort does
* not change the count, and an OR-union needs de-duplication so its count cannot be derived.
*/
private Long computeCoveredCount(FindPlan findPlan, long indexedIdCount) {
if (!findPlan.getSubPlans().isEmpty()
|| findPlan.getCollectionScanFilter() != null
|| findPlan.getSkip() != null
|| findPlan.getLimit() != null
|| findPlan.getByIdFilter() != null) {
return null;
}
if (findPlan.getIndexDescriptor() != null) {
// the index supplied the exact matching id set
return indexedIdCount >= 0 ? indexedIdCount : null;
}
// pure full scan over the whole collection
return nitriteMap.size();
private boolean isCountCovered(FindPlan findPlan) {
return findPlan.getSubPlans().isEmpty()
&& findPlan.getCollectionScanFilter() == null
&& findPlan.getSkip() == null
&& findPlan.getLimit() == null
&& findPlan.getByIdFilter() == null;
}

/**
Expand Down Expand Up @@ -193,7 +201,7 @@ private static Object indexedValue(DBValue dbValue) {
return dbValue == null || dbValue instanceof DBNull ? null : dbValue.getValue();
}

private RecordStream<Pair<NitriteId, Document>> findSuitableStream(FindPlan findPlan, long[] indexedIdCount) {
private RecordStream<Pair<NitriteId, Document>> findSuitableStream(FindPlan findPlan, IndexScan scan) {
RecordStream<Pair<NitriteId, Document>> rawStream;
RecordStream<Pair<NitriteId, Document>> indexSortedStream = null;

Expand All @@ -202,7 +210,7 @@ private RecordStream<Pair<NitriteId, Document>> findSuitableStream(FindPlan find
List<RecordStream<Pair<NitriteId, Document>>> subStreams = new ArrayList<>();
for (FindPlan subPlan : findPlan.getSubPlans()) {
// a sub-plan's own id count cannot answer the union's count (dedup), so discard it
RecordStream<Pair<NitriteId, Document>> suitableStream = findSuitableStream(subPlan, new long[]{ -1 });
RecordStream<Pair<NitriteId, Document>> suitableStream = findSuitableStream(subPlan, new IndexScan());
subStreams.add(suitableStream);
}

Expand Down Expand Up @@ -233,14 +241,21 @@ private RecordStream<Pair<NitriteId, Document>> findSuitableStream(FindPlan find
if (indexDescriptor != null) {
// get optimized filter
NitriteIndexer indexer = nitriteConfig.findIndexer(indexDescriptor.getIndexType());
LinkedHashSet<NitriteId> nitriteIds = indexer.findByFilter(findPlan, nitriteConfig);
RecordStream<NitriteId> idStream = indexer.findByFilterStream(findPlan, nitriteConfig);
if (idStream != null) {
// the index walks its matches lazily; a size() is counted from it on demand
scan.lazyStream = new IndexedStream(idStream, nitriteMap);
rawStream = scan.lazyStream;
} else {
LinkedHashSet<NitriteId> nitriteIds = indexer.findByFilter(findPlan, nitriteConfig);

// the index supplied the exact matching id set; record its size so a size()
// with no row-dropping step downstream can answer from it without fetching
indexedIdCount[0] = nitriteIds.size();
// the index supplied the exact matching id set; record its size so a size()
// with no row-dropping step downstream can answer from it without fetching
scan.idCount = nitriteIds.size();

// create indexed stream from optimized filter
rawStream = new IndexedStream(nitriteIds, nitriteMap);
// create indexed stream from optimized filter
rawStream = new IndexedStream(nitriteIds, nitriteMap);
}
} else {
indexSortedStream = indexSortedStream(findPlan);
rawStream = indexSortedStream != null ? indexSortedStream : nitriteMap.entries();
Expand Down
12 changes: 12 additions & 0 deletions nitrite/src/main/java/org/dizitart/no2/common/streams/DocumentStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@

import java.util.Collections;
import java.util.Iterator;
import java.util.function.LongSupplier;

/**
* @since 4.0
Expand All @@ -53,6 +54,13 @@ public class DocumentStream implements DocumentCursor {
@Setter
private Long coveredCount;

/**
* Answers {@link #size()} from the index on demand when the match count is known to be
* covered but the ids are streamed lazily rather than materialized; evaluated once.
*/
@Setter
private LongSupplier coveredCountSupplier;

public DocumentStream(RecordStream<Pair<NitriteId, Document>> recordStream,
ProcessorChain processorChain) {
this.recordStream = recordStream;
Expand All @@ -64,6 +72,10 @@ public long size() {
if (coveredCount != null) {
return coveredCount;
}
if (coveredCountSupplier != null) {
coveredCount = coveredCountSupplier.getAsLong();
return coveredCount;
}
return Iterables.size(this);
}

Expand Down
19 changes: 16 additions & 3 deletions nitrite/src/main/java/org/dizitart/no2/common/streams/IndexedStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,17 +24,16 @@
import org.dizitart.no2.store.NitriteMap;

import java.util.Iterator;
import java.util.Set;

/**
* @author Anindya Chatterjee
* @since 4.0
*/
public class IndexedStream implements RecordStream<Pair<NitriteId, Document>> {
private final NitriteMap<NitriteId, Document> nitriteMap;
private final Set<NitriteId> nitriteIds;
private final Iterable<NitriteId> nitriteIds;

public IndexedStream(Set<NitriteId> nitriteIds,
public IndexedStream(Iterable<NitriteId> nitriteIds,
NitriteMap<NitriteId, Document> nitriteMap) {
this.nitriteIds = nitriteIds;
this.nitriteMap = nitriteMap;
Expand All @@ -45,6 +44,20 @@ public Iterator<Pair<NitriteId, Document>> iterator() {
return new IndexedStreamIterator(nitriteIds.iterator(), nitriteMap);
}

/**
* Counts the ids the index supplied, walking the id source only, without fetching a
* single document.
*
* @return the number of ids
*/
public long countIds() {
long count = 0;
for (NitriteId ignored : nitriteIds) {
count++;
}
return count;
}

private static class IndexedStreamIterator implements Iterator<Pair<NitriteId, Document>>,
SkippableIterator {
private final Iterator<NitriteId> iterator;
Expand Down
7 changes: 7 additions & 0 deletions nitrite/src/main/java/org/dizitart/no2/index/ComparableIndexer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.dizitart.no2.collection.NitriteId;
import org.dizitart.no2.common.DBValue;
import org.dizitart.no2.common.FieldValues;
import org.dizitart.no2.common.RecordStream;
import org.dizitart.no2.common.Fields;
import org.dizitart.no2.common.tuples.Pair;
import org.dizitart.no2.exceptions.IndexingException;
Expand Down Expand Up @@ -66,6 +67,12 @@ public LinkedHashSet<NitriteId> findByFilter(FindPlan findPlan, NitriteConfig ni
return nitriteIndex.findNitriteIds(findPlan);
}

@Override
public RecordStream<NitriteId> findByFilterStream(FindPlan findPlan, NitriteConfig nitriteConfig) {
NitriteIndex nitriteIndex = findNitriteIndex(findPlan.getIndexDescriptor(), nitriteConfig);
return nitriteIndex.findNitriteIdStream(findPlan);
}

@Override
public List<Pair<DBValue, NitriteId>> readSortKeys(IndexDescriptor indexDescriptor,
NitriteConfig nitriteConfig,
Expand Down
14 changes: 14 additions & 0 deletions nitrite/src/main/java/org/dizitart/no2/index/NitriteIndex.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.dizitart.no2.collection.NitriteId;
import org.dizitart.no2.common.DBValue;
import org.dizitart.no2.common.FieldValues;
import org.dizitart.no2.common.RecordStream;
import org.dizitart.no2.common.tuples.Pair;
import org.dizitart.no2.exceptions.UniqueConstraintException;
import org.dizitart.no2.exceptions.ValidationException;
Expand Down Expand Up @@ -74,6 +75,19 @@ public interface NitriteIndex {
*/
LinkedHashSet<NitriteId> findNitriteIds(FindPlan findPlan);

/**
* Streams the ids matching the plan lazily, in index order and without duplicates, or
* returns {@code null} when this index cannot do so for the given plan, in which case the
* caller falls back to {@link #findNitriteIds(FindPlan)}. A stream lets a query that only
* needs the first rows, or a bounded page, stop reading the index as soon as it has them.
*
* @param findPlan the find plan
* @return a re-iterable stream of ids, or {@code null}
*/
default RecordStream<NitriteId> findNitriteIdStream(FindPlan findPlan) {
return null;
}

/**
* Reads every {@code (indexed value, id)} pair out of the index, so a sorted query can
* decide its order without deserializing a single document.
Expand Down
14 changes: 14 additions & 0 deletions nitrite/src/main/java/org/dizitart/no2/index/NitriteIndexer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.dizitart.no2.collection.NitriteId;
import org.dizitart.no2.common.DBValue;
import org.dizitart.no2.common.FieldValues;
import org.dizitart.no2.common.RecordStream;
import org.dizitart.no2.common.Fields;
import org.dizitart.no2.common.module.NitritePlugin;
import org.dizitart.no2.common.tuples.Pair;
Expand Down Expand Up @@ -88,6 +89,19 @@ public interface NitriteIndexer extends NitritePlugin {
*/
LinkedHashSet<NitriteId> findByFilter(FindPlan findPlan, NitriteConfig nitriteConfig);

/**
* Streams the ids matching the plan lazily, or returns {@code null} when the indexer has no
* lazy path for it and {@link #findByFilter(FindPlan, NitriteConfig)} must be used. The
* default is {@code null}, so existing indexer plugins are unaffected.
*
* @param findPlan the find plan
* @param nitriteConfig the nitrite config
* @return a re-iterable stream of ids, or {@code null}
*/
default RecordStream<NitriteId> findByFilterStream(FindPlan findPlan, NitriteConfig nitriteConfig) {
return null;
}

/**
* Reads every {@code (indexed value, id)} pair out of the given index, so a sorted query
* can decide its order without deserializing a single document.
Expand Down
Loading
Loading