This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 4b7b358b9dd Push AND restrictions into composite filter children
(#19408)
4b7b358b9dd is described below
commit 4b7b358b9dd9051ba238193ebe05d29c0da3cbf0
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Fri Oct 9 19:15:51 2026 +0200
Push AND restrictions into composite filter children (#19408)
<!-- pinot-pr-flow:start -->
### PR flow
Pushes AND restriction down to composite children and scans when enabled\.
```mermaid
flowchart TD
N0["InstancePlanMakerImplV2#46;applyQueryOptions #40;F18#41;"]:::stAdded
N1["QueryContext#46;isAndRestrictionPushdownEnabled
#40;F20#41;"]:::stAdded
N2["AndFilterOperator constructor #40;F13#41;"]:::stModified
N3["AndDocIdSet constructor #40;F4#41;"]:::stModified
N4["AndDocIdSet#46;iterator#40;#41; #40;F4#41;"]:::stModified
N5["AndDocIdSet#46;applyAnd#40;#41; #40;F4#41;"]:::stAdded
N6["OrDocIdSet#46;applyAnd#40;#41; #40;F10#41;"]:::stModified
N7["NotDocIdSet#46;applyAnd#40;#41; #40;F9#41;"]:::stModified
N8["ScanBasedDocIdIterator#46;applyAnd#40;#41; #40;F3#41;"]:::stModified
N0 -->|"sets flag from config#47;query option to QueryContext field"| N1
N1 -->|"AndFilterOperator reads flag via getter"| N2
N2 -->|"AndFilterOperator passes flag to AndDocIdSet constructor"| N3
N3 -->|"AndDocIdSet#46;iterator checks #95;restrictionPushdownEnabled to
#40;"| N4
classDef stAdded
fill:#dafbe1,stroke:#1a7f37,color:#1f2328,stroke-width:2px
classDef stModified
fill:#fff8c5,stroke:#9a6700,color:#1f2328,stroke-width:2px
classDef stRemoved
fill:#ffebe9,stroke:#cf222e,color:#1f2328,stroke-width:2px
classDef stUnchanged
fill:#f6f8fa,stroke:#656d76,color:#1f2328,stroke-width:1px
```
AI-generated · Green: added · Yellow: modified · Red: removed · Gray:
existing
<details>
<summary>Diff evidence</summary>
- F3:
pinot\-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator.java)
- F4:
pinot\-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet.java)
- F9:
pinot\-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet.java)
- F10:
pinot\-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet.java)
- F13:
pinot\-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator.java)
- F18:
pinot\-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java)
- F20:
pinot\-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext\.java
—
[before](https://github.com/apache/pinot/blob/15b838a5df8d1cc34ab6302289defed31c7f44f9/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java)
·
[after](https://github.com/apache/pinot/blob/0c87bb5ae77c1c08707ed05eede55567acf5447e/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java)
</details>
- [ ] Regenerate PR flow
<!-- pinot-pr-flow:meta
eyJiYXNlX3NoYSI6IjE1YjgzOGE1ZGY4ZDFjYzM0YWI2MzAyMjg5ZGVmZWQzMWM3ZjQ0ZjkiLCJjb250ZXh0X2hhc2giOiJkYjdhODRhM2NkMTU3MmYxYmY5ZGRjNjE2ZmExYjJiYmU5Y2NhMzAxOGI4MjgyMzMxZDU5ZDU4YzAyMmEzODg5IiwiZ2VuZXJhdGVkX2F0IjoxNzkwNjczMTQ3LCJoZWFkX3NoYSI6IjBjODdiYjVhZTc3YzFjMDg3MDdlZDA1ZWVkZTU1NTY3YWNmNTQ0N2UiLCJtb2RlbCI6Im52aWRpYS9uZW1vdHJvbi0zLXN1cGVyLTEyMGItYTEyYjpmcmVlIiwib2JzZXJ2ZWRfYmFzZV9zaGEiOiJlZjc1ZWQ5NzE0OGY1ZWIyOTRlNTQ0YjkxODVhMzQ4MzdjZDViMzU4IiwicG9saWN5IjoicGlub3QtcHItZ
[...]
<!-- pinot-pr-flow:signature
3194e019799374a8f9b31d0b2e0cc39757184be42c3cbdcbe0c57351222b504c -->
<!-- pinot-pr-flow:end -->
Fixes #19339
## Problem
When a filter has the shape `P AND (A OR B)`, the `(A OR B)` subtree is
evaluated independently of `P`.
`AndDocIdSet#iterator()` is the only place a restriction is ever pushed
anywhere: it merges the index-based children into a bitmap and hands it to each
`ScanBasedDocIdIterator` via `applyAnd`. `OrDocIdSet` has no equivalent, so a
predicate nested inside an OR branch never learns about `P`.
That has two consequences:
- A predicate with no usable index inside an OR branch — `IN_SUBQUERY`
evaluated through `ExpressionScanDocIdIterator`, or a plain scan on a segment
missing its range index — is evaluated on every document its branch matches,
however selective `P` is.
- **Adding an index to a column inside an OR branch can make a query much
slower.** With no index-based sibling the branch stays lazy and is driven by
the outer merged bitmap; add one and the branch takes the eager path in
`AndDocIdSet#iterator()`, materializing over the whole segment before the outer
AND intersects. In one deployment the same query returning the same two rows
went from ~500 to ~17,000,000 `numEntriesScannedInFilter`, allocating 4.6 GB,
after two columns in the branch [...]
The same gap has been reported twice before and both reports are still
open: #10396 and #9402.
## Fix
Generalize `applyAnd` from `ScanBasedDocIdIterator` up to `BlockDocIdSet`,
so an AND can hand the document ids it has matched to its composite children:
- `AndDocIdSet#applyAnd` adds the candidate set to its children as one more
index-based child and runs `iterator()`. Everything `iterator()` already does —
the sorted-range intersection, sorting bitmaps by cardinality,
`AndScanReordering`, running every scan against the merged document ids —
therefore applies to the restricted evaluation too, with a single
implementation of AND.
- `OrDocIdSet#applyAnd` unions the branches, each restricted to the
candidates no earlier branch has matched, using `(A OR B) AND S == (A AND S) OR
(B AND (S MINUS (A AND S)))`. A scan-based branch costs time linear in the
candidate set it is handed, so narrowing avoids one full pass per branch.
- `NotDocIdSet#applyAnd` returns the candidates the child does not match,
so the child is only evaluated on the candidates.
`AndDocIdSet#iterator()` defers `iterator()` on children that report
`isApplyAndDeferrable()` and pushes its merged document ids into them instead.
With no index-based child to seed the push-down it falls back to the lazy
`AndDocIdIterator`, exactly as before.
Deferral is limited to subtrees that still have a scan to restrict, via
`BlockDocIdSet#isScanBased()`, which the composites compute over their
children. An index-only subtree produces the same bitmap either way, so handing
it a candidate set would only add an intersection per branch. Reading this off
the DocIdSet rather than the operator is also more accurate: an operator that
scans internally and returns a bitmap — a non-exact range index, an H3 index —
has already paid for the scan [...]
## This ships disabled
The push-down **materializes** the filter result instead of streaming it,
so a query that stops before reading every matching document ends up doing the
full filter work. It is gated by `AndRestrictionPushdownMode`, defaulting to
`NEVER`:
| Mode | Behavior |
|---|---|
| `NEVER` (default) | Never push down. Provably identical to today. |
| `ALWAYS` | Always push down. |
| `AUTO` | Push down only for queries that read every matching document —
aggregation and group-by. |
`AUTO` excludes a selection query without ORDER BY (stops at its LIMIT), a
DISTINCT query (`DistinctOperator` stops the same way, and DISTINCT is modelled
as an aggregation function so it needs excluding explicitly), `LIMIT 0`, and —
conservatively — a selection query *with* ORDER BY, because whether it can stop
early is decided per segment in `SelectionPlanNode`, which picks
`SelectionPartiallyOrderedBy*` from the segment's sorted column.
- Server config: `pinot.server.query.executor.and.restriction.pushdown.mode`
- Query option: `SET andRestrictionPushdownMode = 'always'`
The mode is resolved once per query in
`InstancePlanMakerImplV2#applyQueryOptions` and carried on `QueryContext`, so
the filter operators only ever see the resolved flag.
## Benchmark
`BenchmarkAndRestrictionPushdown` (new, in `pinot-perf`) measures
wall-clock time, and allocation under `-prof gc`, because
`numEntriesScannedInFilter` can fall while both of those rise. 2M documents,
real `SVScanDocIdIterator` over a fixed-bit forward index. Ratios of push-down
on to push-down off — lower is better:
| shape | consume | sel | time | alloc |
|---|---|---|---|---|
| SCAN_IN_OR | DRAIN | 0.01 | **0.03x** | 0.10x |
| SCAN_IN_OR | DRAIN | 0.5 | 0.67x | 1.14x |
| SCAN_IN_OR | LIMIT | 0.01 | **0.02x** | 0.10x |
| SCAN_IN_OR | LIMIT | 0.5 | 0.64x | 1.14x |
| SCAN_ONLY_OR | DRAIN | 0.01 | 0.28x | 21.95x |
| SCAN_ONLY_OR | DRAIN | 0.5 | 0.34x | 231.14x |
| SCAN_ONLY_OR | LIMIT | 0.01 | **144.77x** | 25.04x |
| SCAN_ONLY_OR | LIMIT | 0.5 | **9431.50x** | 383.65x |
| INDEXED_OR | all four | | 0.99–1.02x | 1.00x |
`SCAN_IN_OR` is a scan next to an index-based predicate inside an OR
branch, the reported shape. `SCAN_ONLY_OR` is scans directly under the OR with
no index-based sibling. `INDEXED_OR` has no scan anywhere.
Three things to take from it:
- **`INDEXED_OR` is neutral**, which is what restricting deferral to
scan-bearing subtrees is for.
- **The reported shape gains most where the enclosing AND is selective** —
30 to 50 times — which is the production case.
- **`SCAN_ONLY_OR` with a LIMIT is the shape `AUTO` exists to exclude, and
it is far worse than an entries-scanned proxy suggests.** That OR is already
lazy on master, so materializing it replaces a scan of a few dozen documents
with a scan of the whole segment. This is the reason the default is `NEVER`,
and it means `ALWAYS` is genuinely dangerous on a selection workload — it is
not a "slightly more work" setting.
## Follow-up
Delivering the restriction as a stream rather than a bitmap removes that
cliff: worst case 3.27x instead of 9431x, allocation 1.33x instead of 384x.
Spike with numbers in #19589. It does not remove the need for a mode — eager is
still the better strategy for a full drain — but it changes what the mode
chooses, and makes the excluded branch fast rather than merely safe.
Pushing the restriction further down, into the range index itself via
`RangeBitmap`'s context parameter (#7597, and @richardstartin's suggestion on
the issue), is a separate layer again and is not addressed here.
## Test plan
- `AndDocIdSetPushdownTest` — the reported shape (`AND(P, OR(AND(idx,
scan), other))`): the scan visits 100 documents instead of 5000, with identical
results. Plus the NOT case, the no-index-child fallback, sorted-range
boundaries, the `numDocs` bound on NOT and MATCH-ALL, release of
short-circuited OR branches, deferral limited to scan-bearing subtrees, and 200
randomized filter trees over bitmap/scan/sorted/match-all/empty leaves
asserting identical document ids with the push-down o [...]
- `AndRestrictionPushdownModeTest` — mode resolution: every `AUTO`
inclusion and exclusion, `ALWAYS`/`NEVER`, case-insensitivity, invalid values
rejected, server default, and the query option overriding it.
- `AndRestrictionPushdownQueriesTest` — end to end against a real segment
in every mode: `ALWAYS` scans strictly fewer entries than `NEVER` with
identical rows and `numDocsScanned`; `AUTO` matches `NEVER` for selections and
`ALWAYS` for aggregations; the default is pinned to off. Covers filtered
aggregations (`CombinedFilterOperator`) and null handling, where `getFalses()`
wraps the push-down in a NOT over `OR(trues, nulls)`.
- Full `org.apache.pinot.core.**` and `org.apache.pinot.queries.**` suites
pass. No existing expectation changed, because the default leaves every
existing query on its current plan.
## Release notes
Two new user-facing configuration surfaces, both defaulting to the current
behavior: the server config
`pinot.server.query.executor.and.restriction.pushdown.mode` and the query
option `andRestrictionPushdownMode`. Enabling either changes
`numEntriesScannedInFilter` for affected queries. Old servers in a
mixed-version cluster ignore the unknown query option, which matches the
`NEVER` default.
---
.../common/utils/config/QueryOptionsUtils.java | 21 +
.../apache/pinot/core/common/BlockDocIdSet.java | 95 ++++-
.../dociditerators/MVScanDocIdIterator.java | 55 +--
.../pinot/core/operator/docidsets/AndDocIdSet.java | 121 +++++-
.../core/operator/docidsets/EmptyDocIdSet.java | 7 +
.../operator/docidsets/ExpressionDocIdSet.java | 5 +
.../core/operator/docidsets/MVScanDocIdSet.java | 5 +
.../core/operator/docidsets/MatchAllDocIdSet.java | 7 +
.../pinot/core/operator/docidsets/NotDocIdSet.java | 47 +++
.../pinot/core/operator/docidsets/OrDocIdSet.java | 90 +++-
.../core/operator/docidsets/SVScanDocIdSet.java | 5 +
.../core/operator/filter/AndFilterOperator.java | 12 +-
.../operator/filter/BaseColumnFilterOperator.java | 5 +-
.../operator/filter/CombinedFilterOperator.java | 7 +-
.../core/operator/filter/FilterOperatorUtils.java | 2 +-
.../core/operator/filter/MapFilterOperator.java | 7 +-
.../core/plan/maker/InstancePlanMakerImplV2.java | 41 ++
.../function/AggregationFunctionUtils.java | 3 +-
.../core/query/request/context/QueryContext.java | 14 +
.../docidsets/AndDocIdSetPushdownTest.java | 465 +++++++++++++++++++++
.../plan/maker/AndRestrictionPushdownModeTest.java | 129 ++++++
.../AndRestrictionPushdownNullQueriesTest.java | 201 +++++++++
.../queries/AndRestrictionPushdownQueriesTest.java | 109 +++++
.../perf/BenchmarkAndRestrictionPushdown.java | 208 +++++++++
.../apache/pinot/spi/utils/CommonConstants.java | 25 ++
25 files changed, 1636 insertions(+), 50 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
index f5f8168ca5a..68b32d3b122 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
@@ -39,6 +39,7 @@ import org.apache.pinot.spi.utils.CommonConstants;
import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import
org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner.JoinOverFlowMode;
import
org.apache.pinot.spi.utils.CommonConstants.MultiStageQueryRunner.WindowOverFlowMode;
+import
org.apache.pinot.spi.utils.CommonConstants.Server.AndRestrictionPushdownMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -342,6 +343,26 @@ public class QueryOptionsUtils {
return
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.AND_SCAN_REORDERING));
}
+ /// Per-query override of [AndRestrictionPushdownMode], or `null` when the
query does not set one, in which case
+ /// the server default applies.
+ @Nullable
+ public static AndRestrictionPushdownMode
getAndRestrictionPushdownMode(Map<String, String> queryOptions) {
+ String mode =
queryOptions.get(QueryOptionKey.AND_RESTRICTION_PUSHDOWN_MODE);
+ return mode != null ?
parseAndRestrictionPushdownMode(QueryOptionKey.AND_RESTRICTION_PUSHDOWN_MODE,
mode) : null;
+ }
+
+ /// Parses an [AndRestrictionPushdownMode], naming the offending key and the
legal values on failure. The option
+ /// reads like a toggle, so `true`/`false` is a likely mistake and the
message has to say so.
+ public static AndRestrictionPushdownMode
parseAndRestrictionPushdownMode(String key, String value) {
+ try {
+ return
AndRestrictionPushdownMode.valueOf(value.toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ String.format("Invalid value for %s: '%s'. Expected one of: %s (this
option is a mode, not a boolean toggle)",
+ key, value,
Arrays.toString(AndRestrictionPushdownMode.values())));
+ }
+ }
+
public static boolean isSkipUpsert(Map<String, String> queryOptions) {
return Boolean.parseBoolean(queryOptions.get(QueryOptionKey.SKIP_UPSERT));
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/common/BlockDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/common/BlockDocIdSet.java
index c960727dd46..f89cb43a56b 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/common/BlockDocIdSet.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/common/BlockDocIdSet.java
@@ -19,16 +19,20 @@
package org.apache.pinot.core.common;
import org.apache.pinot.core.operator.dociditerators.AndDocIdIterator;
+import org.apache.pinot.core.operator.dociditerators.BitmapBasedDocIdIterator;
import org.apache.pinot.core.operator.dociditerators.BitmapDocIdIterator;
import org.apache.pinot.core.operator.dociditerators.EmptyDocIdIterator;
import org.apache.pinot.core.operator.dociditerators.OrDocIdIterator;
import
org.apache.pinot.core.operator.dociditerators.RangelessBitmapDocIdIterator;
import org.apache.pinot.core.operator.dociditerators.ScanBasedDocIdIterator;
+import org.apache.pinot.core.operator.dociditerators.SortedDocIdIterator;
import org.apache.pinot.core.operator.docidsets.BitmapDocIdSet;
import org.apache.pinot.core.operator.docidsets.EmptyDocIdSet;
import org.apache.pinot.core.operator.docidsets.RangelessBitmapDocIdSet;
import org.apache.pinot.segment.spi.Constants;
+import org.apache.pinot.spi.utils.Pairs.IntPair;
import org.roaringbitmap.RoaringBitmapWriter;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
import org.roaringbitmap.buffer.MutableRoaringBitmap;
@@ -48,6 +52,89 @@ public interface BlockDocIdSet {
/// filtering phase. This method should be called after the filtering is
done.
long getNumEntriesScannedInFilter();
+ /// Returns whether evaluating this DocIdSet costs time proportional to the
number of documents it is asked about,
+ /// i.e. whether the subtree still has a scan- or expression-based predicate
left to evaluate.
+ ///
+ /// Only such a subtree gains anything from being handed a candidate set. An
index-only subtree produces the same
+ /// bitmap either way, so restricting it just adds intersection work. Note
this is a property of the *DocIdSet*, not
+ /// of the operator that built it: an operator that scans internally and
hands back a bitmap -- a non-exact range
+ /// index, an H3 index -- has already paid for the scan by the time its
DocIdSet exists, and correctly reports
+ /// `false`.
+ default boolean isScanBased() {
+ return false;
+ }
+
+ /// Returns whether a parent AND may defer calling [#iterator] on this
DocIdSet and reach it through [#applyAnd]
+ /// instead.
+ ///
+ /// True only for a composite DocIdSet (AND, OR, NOT) whose subtree is still
scan-based. Forwarding a candidate set
+ /// to its children is how a restriction from an enclosing AND reaches a
scan-based predicate nested inside an OR,
+ /// which would otherwise be evaluated against every document that branch
matches. A leaf is never deferrable: its
+ /// index lookup has already happened, and a scan-based leaf is already
reached through
+ /// [ScanBasedDocIdIterator#applyAnd] by the enclosing AND.
+ ///
+ /// This says nothing about whether [#applyAnd] may be called -- every
DocIdSet supports it.
+ default boolean isApplyAndDeferrable() {
+ return false;
+ }
+
+ /// Returns the document ids matching this DocIdSet that are also in
`docIds`, i.e. the intersection of this
+ /// DocIdSet with the given candidate set.
+ ///
+ /// `docIds` is never modified, so the same candidate set may be handed to
several DocIdSets. The returned bitmap
+ /// must not be modified by the caller either: it may be a view of a bitmap
the DocIdSet still owns.
+ ///
+ /// Like [ScanBasedDocIdIterator#applyAnd], this method consumes the
DocIdSet: call it at most once, and never
+ /// together with [#iterator]. [#getNumEntriesScannedInFilter] stays valid
afterwards.
+ default ImmutableRoaringBitmap applyAnd(ImmutableRoaringBitmap docIds) {
+ if (docIds.isEmpty()) {
+ // Nothing is evaluated, so the iterator will not close itself
+ release();
+ return new MutableRoaringBitmap();
+ }
+ BlockDocIdIterator docIdIterator = iterator();
+ if (docIdIterator instanceof ScanBasedDocIdIterator) {
+ // The scan only visits the candidate documents, which is the whole
point of the push-down
+ return ((ScanBasedDocIdIterator) docIdIterator).applyAnd(docIds);
+ }
+ if (docIdIterator instanceof BitmapBasedDocIdIterator) {
+ return ImmutableRoaringBitmap.and(((BitmapBasedDocIdIterator)
docIdIterator).getDocIds(), docIds);
+ }
+ if (docIdIterator instanceof SortedDocIdIterator) {
+ MutableRoaringBitmap docIdsFromRanges = new MutableRoaringBitmap();
+ for (IntPair docIdRange : ((SortedDocIdIterator)
docIdIterator).getDocIdRanges()) {
+ // NOTE: docIdRange has inclusive start and end.
+ docIdsFromRanges.add(docIdRange.getLeft(), docIdRange.getRight() + 1L);
+ }
+ docIdsFromRanges.and(docIds);
+ return docIdsFromRanges;
+ }
+ // Generic fallback: drive the iterator from the candidate set so that it
is only asked about candidate documents
+ return collect(new AndDocIdIterator(
+ new BlockDocIdIterator[]{new RangelessBitmapDocIdIterator(docIds),
docIdIterator}));
+ }
+
+ /// Releases a DocIdSet that will not be evaluated, e.g. an OR branch a
short-circuit skipped past.
+ ///
+ /// Scan-based DocIdSets build their iterator in the constructor, and that
iterator holds a
+ /// `ForwardIndexReaderContext` -- a direct ByteBuffer for a
chunk-compressed column. Evaluation normally releases it
+ /// because the iterator closes itself on reaching EOF, so a DocIdSet that
is never evaluated has to be released
+ /// explicitly. Calling this instead of evaluating never reads any data.
+ default void release() {
+ iterator().close();
+ }
+
+ /// Materializes the document ids remaining in the given iterator.
+ static MutableRoaringBitmap collect(BlockDocIdIterator docIdIterator) {
+ RoaringBitmapWriter<MutableRoaringBitmap> bitmapWriter =
+ RoaringBitmapWriter.bufferWriter().runCompress(false).get();
+ int docId;
+ while ((docId = docIdIterator.next()) != Constants.EOF) {
+ bitmapWriter.add(docId);
+ }
+ return bitmapWriter.get();
+ }
+
/// For scan-based FilterBlockDocIdSet, pre-scans the documents and returns
a non-scan-based FilterBlockDocIdSet.
default BlockDocIdSet toNonScanDocIdSet() {
BlockDocIdIterator docIdIterator = iterator();
@@ -55,13 +142,7 @@ public interface BlockDocIdSet {
// TODO: This scan is not counted in the execution stats
if (docIdIterator instanceof ScanBasedDocIdIterator || docIdIterator
instanceof AndDocIdIterator
|| docIdIterator instanceof OrDocIdIterator) {
- RoaringBitmapWriter<MutableRoaringBitmap> bitmapWriter =
- RoaringBitmapWriter.bufferWriter().runCompress(false).get();
- int docId;
- while ((docId = docIdIterator.next()) != Constants.EOF) {
- bitmapWriter.add(docId);
- }
- return RangelessBitmapDocIdSet.create(bitmapWriter.get());
+ return RangelessBitmapDocIdSet.create(collect(docIdIterator));
}
// NOTE: AND and OR DocIdSet might return BitmapBasedDocIdIterator after
processing the iterators. Create a new
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator.java
index f9664b68adf..0dd7f99394e 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/dociditerators/MVScanDocIdIterator.java
@@ -81,34 +81,39 @@ public final class MVScanDocIdIterator implements
ScanBasedDocIdIterator {
@Override
public MutableRoaringBitmap applyAnd(BatchIterator docIdIterator,
OptionalInt firstDoc, OptionalInt lastDoc) {
- if (!docIdIterator.hasNext()) {
- return new MutableRoaringBitmap();
- }
- RoaringBitmapWriter<MutableRoaringBitmap> result;
- if (firstDoc.isPresent() && lastDoc.isPresent()) {
- result = RoaringBitmapWriter.bufferWriter()
- .expectedRange(firstDoc.getAsInt(), lastDoc.getAsInt())
- .runCompress(false)
- .get();
- } else {
- result = RoaringBitmapWriter.bufferWriter()
- .runCompress(false)
- .get();
- }
- int[] buffer = new int[OPTIMAL_ITERATOR_BATCH_SIZE];
- while (docIdIterator.hasNext()) {
- int limit = docIdIterator.nextBatch(buffer);
- for (int i = 0; i < limit; i++) {
- int nextDocId = buffer[i];
- // TODO: The performance can be improved by batching the docID lookups
similar to how it's done in
- // SVScanDocIdIterator
- boolean doesValueMatch = _valueMatcher.doesValueMatch(nextDocId);
- if (doesValueMatch) {
- result.add(nextDocId);
+ // NOTE: Close the reader context once the candidates are consumed: unlike
next(), this path never reaches EOF
+ try {
+ if (!docIdIterator.hasNext()) {
+ return new MutableRoaringBitmap();
+ }
+ RoaringBitmapWriter<MutableRoaringBitmap> result;
+ if (firstDoc.isPresent() && lastDoc.isPresent()) {
+ result = RoaringBitmapWriter.bufferWriter()
+ .expectedRange(firstDoc.getAsInt(), lastDoc.getAsInt())
+ .runCompress(false)
+ .get();
+ } else {
+ result = RoaringBitmapWriter.bufferWriter()
+ .runCompress(false)
+ .get();
+ }
+ int[] buffer = new int[OPTIMAL_ITERATOR_BATCH_SIZE];
+ while (docIdIterator.hasNext()) {
+ int limit = docIdIterator.nextBatch(buffer);
+ for (int i = 0; i < limit; i++) {
+ int nextDocId = buffer[i];
+ // TODO: The performance can be improved by batching the docID
lookups similar to how it's done in
+ // SVScanDocIdIterator
+ boolean doesValueMatch = _valueMatcher.doesValueMatch(nextDocId);
+ if (doesValueMatch) {
+ result.add(nextDocId);
+ }
}
}
+ return result.get();
+ } finally {
+ close();
}
- return result.get();
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet.java
index 27db4f64709..0e60e537b11 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/AndDocIdSet.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.operator.docidsets;
+import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
@@ -53,18 +54,37 @@ public final class AndDocIdSet implements BlockDocIdSet {
// Keep the scan based BlockDocIdSets to be accessed when collecting query
execution stats
private final AtomicReference<List<BlockDocIdSet>> _scanBasedDocIdSets = new
AtomicReference<>();
private final boolean _cardinalityBasedRankingForScan;
+ private final boolean _restrictionPushdownEnabled;
private List<BlockDocIdSet> _docIdSets;
private volatile long _numEntriesScannedInFilter;
+ /// Creates an AndDocIdSet that does not push its matching document ids into
its composite children.
public AndDocIdSet(List<BlockDocIdSet> docIdSets, @Nullable Map<String,
String> queryOptions) {
+ this(docIdSets, queryOptions, false);
+ }
+
+ /// @param restrictionPushdownEnabled whether the matching document ids may
be handed to the composite (AND/OR/NOT)
+ /// children. Resolved once per query from
the `andRestrictionPushdown` query
+ /// option and the server default, see
`QueryContext`.
+ public AndDocIdSet(List<BlockDocIdSet> docIdSets, @Nullable Map<String,
String> queryOptions,
+ boolean restrictionPushdownEnabled) {
_docIdSets = docIdSets;
_cardinalityBasedRankingForScan =
queryOptions != null &&
QueryOptionsUtils.isAndScanReorderingEnabled(queryOptions);
+ _restrictionPushdownEnabled = restrictionPushdownEnabled;
}
@Override
public BlockDocIdIterator iterator() {
- int numDocIdSets = _docIdSets.size();
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ Preconditions.checkState(docIdSets != null, "iterator() called on an
already consumed AndDocIdSet");
+ return buildIterator(docIdSets);
+ }
+
+ /// Builds the iterator over the given children and consumes this DocIdSet.
[#applyAnd] passes the children with
+ /// the candidate set prepended, which is why they are not read from
`_docIdSets` here.
+ private BlockDocIdIterator buildIterator(List<BlockDocIdSet> docIdSets) {
+ int numDocIdSets = docIdSets.size();
// NOTE: Keep the order of BlockDocIdSets to preserve the order decided
within FilterOperatorUtils.
// TODO: Consider deciding the order based on the stats of
BlockDocIdIterators
BlockDocIdIterator[] allDocIdIterators = new
BlockDocIdIterator[numDocIdSets];
@@ -72,11 +92,21 @@ public final class AndDocIdSet implements BlockDocIdSet {
List<BitmapBasedDocIdIterator> bitmapBasedDocIdIterators = new
ArrayList<>();
List<ScanBasedDocIdIterator> scanBasedDocIdIterators = new ArrayList<>();
List<BlockDocIdIterator> remainingDocIdIterators = new ArrayList<>();
+ // Composite (AND/OR/NOT) children whose evaluation is deferred so that
the merged index document ids can be
+ // pushed into them. Their slot in allDocIdIterators stays null until the
lazy fallback below fills it.
+ List<BlockDocIdSet> restrictableDocIdSets = new ArrayList<>();
long numEntriesScannedForNonScanBasedDocIdSets = 0L;
List<BlockDocIdSet> scanBasedDocIdSets = new ArrayList<>();
for (int i = 0; i < numDocIdSets; i++) {
- BlockDocIdSet docIdSet = _docIdSets.get(i);
+ BlockDocIdSet docIdSet = docIdSets.get(i);
+ if (_restrictionPushdownEnabled && docIdSet.isApplyAndDeferrable()) {
+ // NOTE: Do not call iterator() here. That evaluates the whole subtree
over the segment, which is exactly what
+ // the push-down below avoids.
+ restrictableDocIdSets.add(docIdSet);
+ scanBasedDocIdSets.add(docIdSet);
+ continue;
+ }
BlockDocIdIterator docIdIterator = docIdSet.iterator();
allDocIdIterators[i] = docIdIterator;
if (docIdIterator instanceof SortedDocIdIterator) {
@@ -94,10 +124,12 @@ public final class AndDocIdSet implements BlockDocIdSet {
}
}
- // Set _docIdSets to null so that underlying BlockDocIdSets can be garbage
collected
- _docIdSets = null;
+ // Publish the stats state before dropping the children, so that a
concurrent reader of
+ // getNumEntriesScannedInFilter() never sees it disappear, then set
_docIdSets to null so that the underlying
+ // BlockDocIdSets can be garbage collected
_numEntriesScannedInFilter = numEntriesScannedForNonScanBasedDocIdSets;
_scanBasedDocIdSets.set(scanBasedDocIdSets);
+ _docIdSets = null;
// evaluate the bitmaps in the order of the lowest matching num docIds
comes first, so that we minimize the number
// of containers (range) for comparison from the beginning, as will
minimize the effort of bitmap AND application
@@ -117,8 +149,10 @@ public final class AndDocIdSet implements BlockDocIdSet {
int numBitmapBasedDocIdIterators = bitmapBasedDocIdIterators.size();
int numScanBasedDocIdIterators = scanBasedDocIdIterators.size();
int numRemainingDocIdIterators = remainingDocIdIterators.size();
+ int numRestrictableDocIdSets = restrictableDocIdSets.size();
int numIndexBasedDocIdIterators = numSortedDocIdIterators +
numBitmapBasedDocIdIterators;
- if ((numIndexBasedDocIdIterators > 0 && numScanBasedDocIdIterators > 0) ||
numIndexBasedDocIdIterators > 1) {
+ if ((numIndexBasedDocIdIterators > 0 && (numScanBasedDocIdIterators > 0 ||
numRestrictableDocIdSets > 0))
+ || numIndexBasedDocIdIterators > 1) {
// When there are at least one index-base BlockDocIdIterator
(SortedDocIdIterator or BitmapBasedDocIdIterator)
// and at least one ScanBasedDocIdIterator, or more than one index-based
BlockDocIdIterator, merge them and
// construct a RangelessBitmapDocIdIterator from the merged document
ids. If there is no remaining
@@ -166,6 +200,18 @@ public final class AndDocIdSet implements BlockDocIdSet {
for (ScanBasedDocIdIterator scanBasedDocIdIterator :
scanBasedDocIdIterators) {
docIds = scanBasedDocIdIterator.applyAnd(docIds);
}
+ // Hand the matching document ids to the composite children, so that a
scan-based predicate nested inside one of
+ // them (typically under an OR) is only evaluated on the documents that
can still match this AND.
+ for (int i = 0; i < numRestrictableDocIdSets; i++) {
+ if (docIds.isEmpty()) {
+ // Nothing can match any more, but the remaining children still hold
reader contexts
+ for (int j = i; j < numRestrictableDocIdSets; j++) {
+ restrictableDocIdSets.get(j).release();
+ }
+ break;
+ }
+ docIds = restrictableDocIdSets.get(i).applyAnd(docIds);
+ }
RangelessBitmapDocIdIterator rangelessBitmapDocIdIterator = new
RangelessBitmapDocIdIterator(docIds);
if (numRemainingDocIdIterators == 0) {
return rangelessBitmapDocIdIterator;
@@ -179,11 +225,64 @@ public final class AndDocIdSet implements BlockDocIdSet {
}
} else {
// Otherwise, construct and return an AndDocIdIterator with all
BlockDocIdIterators.
-
+ // There is no index-based document id set to seed the push-down with,
so evaluate the deferred composite
+ // children now and let AndDocIdIterator drive them lazily, as it did
before the push-down existed.
+ for (int i = 0; i < numDocIdSets; i++) {
+ if (allDocIdIterators[i] == null) {
+ allDocIdIterators[i] = docIdSets.get(i).iterator();
+ }
+ }
return new AndDocIdIterator(allDocIdIterators);
}
}
+ @Override
+ public boolean isScanBased() {
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ if (docIdSets == null) {
+ return false;
+ }
+ for (BlockDocIdSet docIdSet : docIdSets) {
+ if (docIdSet.isScanBased()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean isApplyAndDeferrable() {
+ return isScanBased();
+ }
+
+ /// Intersects this AND with the candidate set by handing the candidates to
[#buildIterator] as one more
+ /// index-based child. Everything [#iterator] does -- merging the index
children first, sorting bitmaps by
+ /// cardinality,
[org.apache.pinot.common.utils.config.QueryOptionsUtils#isAndScanReorderingEnabled],
running every
+ /// scan against the merged document ids -- then applies to the restricted
evaluation too, with one implementation of
+ /// AND.
+ @Override
+ public ImmutableRoaringBitmap applyAnd(ImmutableRoaringBitmap docIds) {
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ Preconditions.checkState(docIdSets != null, "applyAnd() called on an
already consumed AndDocIdSet");
+ if (docIds.isEmpty()) {
+ // No child is evaluated, so none of them will close its own iterator
+ for (BlockDocIdSet docIdSet : docIdSets) {
+ docIdSet.release();
+ }
+ _scanBasedDocIdSets.set(docIdSets);
+ _docIdSets = null;
+ return new MutableRoaringBitmap();
+ }
+ List<BlockDocIdSet> docIdSetsWithCandidates = new
ArrayList<>(docIdSets.size() + 1);
+ docIdSetsWithCandidates.add(RangelessBitmapDocIdSet.create(docIds));
+ docIdSetsWithCandidates.addAll(docIdSets);
+ BlockDocIdIterator docIdIterator = buildIterator(docIdSetsWithCandidates);
+ // buildIterator() returns the merged document ids directly whenever it
has no lazy child left, which is the usual
+ // case once the candidate set has forced the merge branch
+ return docIdIterator instanceof BitmapBasedDocIdIterator ?
((BitmapBasedDocIdIterator) docIdIterator).getDocIds()
+ : BlockDocIdSet.collect(docIdIterator);
+ }
+
@Override
public long getNumEntriesScannedInFilter() {
List<BlockDocIdSet> scanBasedDocIdSets = _scanBasedDocIdSets.get();
@@ -195,4 +294,14 @@ public final class AndDocIdSet implements BlockDocIdSet {
}
return _numEntriesScannedInFilter + numEntriesScannedForScanBasedDocIdSets;
}
+
+ @Override
+ public void release() {
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ if (docIdSets != null) {
+ for (BlockDocIdSet docIdSet : docIdSets) {
+ docIdSet.release();
+ }
+ }
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/EmptyDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/EmptyDocIdSet.java
index ea08723cd8d..ae46e957081 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/EmptyDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/EmptyDocIdSet.java
@@ -20,6 +20,8 @@ package org.apache.pinot.core.operator.docidsets;
import org.apache.pinot.core.common.BlockDocIdSet;
import org.apache.pinot.core.operator.dociditerators.EmptyDocIdIterator;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
/// A [BlockDocIdSet] that does not contain any document.
@@ -49,4 +51,9 @@ public final class EmptyDocIdSet implements BlockDocIdSet {
public long getNumEntriesScannedInFilter() {
return _numEntriesScannedInFilter;
}
+
+ @Override
+ public ImmutableRoaringBitmap applyAnd(ImmutableRoaringBitmap docIds) {
+ return new MutableRoaringBitmap();
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/ExpressionDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/ExpressionDocIdSet.java
index 333a7ab14d3..db4610b1135 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/ExpressionDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/ExpressionDocIdSet.java
@@ -43,6 +43,11 @@ public final class ExpressionDocIdSet implements
BlockDocIdSet {
return _docIdIterator;
}
+ @Override
+ public boolean isScanBased() {
+ return true;
+ }
+
@Override
public long getNumEntriesScannedInFilter() {
return _docIdIterator.getNumEntriesScanned();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MVScanDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MVScanDocIdSet.java
index f6807b79074..4ed5c18ced9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MVScanDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MVScanDocIdSet.java
@@ -38,6 +38,11 @@ public final class MVScanDocIdSet implements BlockDocIdSet {
return _docIdIterator;
}
+ @Override
+ public boolean isScanBased() {
+ return true;
+ }
+
@Override
public long getNumEntriesScannedInFilter() {
return _docIdIterator.getNumEntriesScanned();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MatchAllDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MatchAllDocIdSet.java
index 497be02bd4f..d971982aebc 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MatchAllDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/MatchAllDocIdSet.java
@@ -20,6 +20,7 @@ package org.apache.pinot.core.operator.docidsets;
import org.apache.pinot.core.common.BlockDocIdSet;
import org.apache.pinot.core.operator.dociditerators.MatchAllDocIdIterator;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
public final class MatchAllDocIdSet implements BlockDocIdSet {
@@ -38,4 +39,10 @@ public final class MatchAllDocIdSet implements BlockDocIdSet
{
public long getNumEntriesScannedInFilter() {
return 0L;
}
+
+ @Override
+ public ImmutableRoaringBitmap applyAnd(ImmutableRoaringBitmap docIds) {
+ // MatchAllDocIdIterator stops at numDocs, so the candidate set has to be
bounded the same way
+ return NotDocIdSet.bound(docIds, _numDocs);
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet.java
index d44b263c74a..65fdbe4ee07 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/NotDocIdSet.java
@@ -21,6 +21,8 @@ package org.apache.pinot.core.operator.docidsets;
import org.apache.pinot.core.common.BlockDocIdIterator;
import org.apache.pinot.core.common.BlockDocIdSet;
import org.apache.pinot.core.operator.dociditerators.NotDocIdIterator;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
public final class NotDocIdSet implements BlockDocIdSet {
@@ -41,4 +43,49 @@ public final class NotDocIdSet implements BlockDocIdSet {
public long getNumEntriesScannedInFilter() {
return _childDocIdSet.getNumEntriesScannedInFilter();
}
+
+ @Override
+ public boolean isScanBased() {
+ return _childDocIdSet.isScanBased();
+ }
+
+ @Override
+ public boolean isApplyAndDeferrable() {
+ return isScanBased();
+ }
+
+ @Override
+ public void release() {
+ _childDocIdSet.release();
+ }
+
+ @Override
+ public ImmutableRoaringBitmap applyAnd(ImmutableRoaringBitmap docIds) {
+ if (docIds.isEmpty()) {
+ // The child is not evaluated, so it will not close its own iterator
+ _childDocIdSet.release();
+ return new MutableRoaringBitmap();
+ }
+ // NOTE: The candidate set can carry document ids beyond numDocs, because
BitmapDocIdIterator#getDocIds() hands out
+ // the raw bitmap while its next() stops at numDocs. An intersection
cannot introduce such ids, but a
+ // complement can, so the bound NotDocIdIterator applies has to be
applied here too.
+ ImmutableRoaringBitmap boundedDocIds = bound(docIds, _numDocs);
+ if (boundedDocIds.isEmpty()) {
+ _childDocIdSet.release();
+ return new MutableRoaringBitmap();
+ }
+ // Within the candidate set, NOT(child) is the candidates the child does
not match, so the child only has to be
+ // evaluated on the candidates: (NOT child) AND docIds == docIds MINUS
(child AND docIds).
+ return ImmutableRoaringBitmap.andNot(boundedDocIds,
_childDocIdSet.applyAnd(boundedDocIds));
+ }
+
+ /// Drops the document ids that are not below `numDocs`, without copying
when there are none.
+ static ImmutableRoaringBitmap bound(ImmutableRoaringBitmap docIds, int
numDocs) {
+ if (docIds.isEmpty() || docIds.last() < numDocs) {
+ return docIds;
+ }
+ MutableRoaringBitmap boundedDocIds = docIds.toMutableRoaringBitmap();
+ boundedDocIds.remove(numDocs, 0x1_0000_0000L);
+ return boundedDocIds;
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet.java
index 2705a6d4e28..0ce14c22519 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/OrDocIdSet.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.operator.docidsets;
+import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
@@ -28,6 +29,7 @@ import
org.apache.pinot.core.operator.dociditerators.BitmapDocIdIterator;
import org.apache.pinot.core.operator.dociditerators.OrDocIdIterator;
import org.apache.pinot.core.operator.dociditerators.SortedDocIdIterator;
import org.apache.pinot.spi.utils.Pairs;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
import org.roaringbitmap.buffer.MutableRoaringBitmap;
@@ -55,6 +57,7 @@ public final class OrDocIdSet implements BlockDocIdSet {
@Override
public BlockDocIdIterator iterator() {
+ Preconditions.checkState(_docIdSets != null, "iterator() called on an
already consumed OrDocIdSet");
int numDocIdSets = _docIdSets.size();
BlockDocIdIterator[] allDocIdIterators = new
BlockDocIdIterator[numDocIdSets];
List<SortedDocIdIterator> sortedDocIdIterators = new ArrayList<>();
@@ -79,10 +82,12 @@ public final class OrDocIdSet implements BlockDocIdSet {
}
}
- // Set _docIdSets to null so that underlying BlockDocIdSets can be garbage
collected
- _docIdSets = null;
+ // Publish the stats state before dropping the branches, so that a
concurrent reader of
+ // getNumEntriesScannedInFilter() never sees it disappear, then set
_docIdSets to null so that the underlying
+ // BlockDocIdSets can be garbage collected
_numEntriesScannedInFilter = numEntriesScannedForNonScanBasedDocIdSets;
_scanBasedDocIdSets.set(scanBasedDocIdSets);
+ _docIdSets = null;
int numSortedDocIdIterators = sortedDocIdIterators.size();
int numBitmapBasedDocIdIterators = bitmapBasedDocIdIterators.size();
@@ -132,4 +137,85 @@ public final class OrDocIdSet implements BlockDocIdSet {
}
return _numEntriesScannedInFilter + numEntriesScannedForScanBasedDocIdSets;
}
+
+ @Override
+ public boolean isScanBased() {
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ if (docIdSets == null) {
+ return false;
+ }
+ for (BlockDocIdSet docIdSet : docIdSets) {
+ if (docIdSet.isScanBased()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public boolean isApplyAndDeferrable() {
+ return isScanBased();
+ }
+
+ /// Unions the branches, each restricted to the candidate document ids.
+ ///
+ /// A branch only has to look at the candidates no earlier branch has
matched yet, because
+ /// `(A OR B) AND S == (A AND S) OR (B AND (S MINUS (A AND S)))`. That
matters because the cost of a scan-based
+ /// branch is linear in the size of the candidate set it is given, so
handing every branch the full candidate set
+ /// would scan it once per branch.
+ @Override
+ public ImmutableRoaringBitmap applyAnd(ImmutableRoaringBitmap docIds) {
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ Preconditions.checkState(docIdSets != null, "applyAnd() called on an
already consumed OrDocIdSet");
+ // Publish the stats state before dropping the branches, so that a
concurrent reader of
+ // getNumEntriesScannedInFilter() never sees it disappear. Branches left
unevaluated by the short-circuit below
+ // report zero.
+ _scanBasedDocIdSets.set(docIdSets);
+ _docIdSets = null;
+ MutableRoaringBitmap docIdsToReturn = new MutableRoaringBitmap();
+ // NOTE: Bound the candidates like NotDocIdSet does. A bitmap branch is
intersected through its raw bitmap, which
+ // on a realtime segment can carry document ids beyond numDocs that
the iterator() path would have dropped.
+ ImmutableRoaringBitmap boundedDocIds = NotDocIdSet.bound(docIds, _numDocs);
+ if (boundedDocIds.isEmpty()) {
+ // No branch is evaluated, so none of them will close its own iterator
+ releaseFrom(docIdSets, 0);
+ return docIdsToReturn;
+ }
+ int numCandidates = boundedDocIds.getCardinality();
+ int numDocIdSets = docIdSets.size();
+ ImmutableRoaringBitmap remainingDocIds = boundedDocIds;
+ for (int i = 0; i < numDocIdSets; i++) {
+ ImmutableRoaringBitmap matchingDocIds =
docIdSets.get(i).applyAnd(remainingDocIds);
+ if (matchingDocIds.isEmpty()) {
+ continue;
+ }
+ docIdsToReturn.or(matchingDocIds);
+ if (docIdsToReturn.getCardinality() == numCandidates) {
+ // Every candidate is already matched, so the remaining branches
cannot add anything. They still have to be
+ // released: their iterators were built eagerly and hold forward-index
reader contexts.
+ releaseFrom(docIdSets, i + 1);
+ break;
+ }
+ if (i < numDocIdSets - 1) {
+ remainingDocIds = ImmutableRoaringBitmap.andNot(remainingDocIds,
matchingDocIds);
+ }
+ }
+ return docIdsToReturn;
+ }
+
+ private static void releaseFrom(List<BlockDocIdSet> docIdSets, int
fromIndex) {
+ for (int i = fromIndex; i < docIdSets.size(); i++) {
+ docIdSets.get(i).release();
+ }
+ }
+
+ @Override
+ public void release() {
+ List<BlockDocIdSet> docIdSets = _docIdSets;
+ if (docIdSets != null) {
+ for (BlockDocIdSet docIdSet : docIdSets) {
+ docIdSet.release();
+ }
+ }
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/SVScanDocIdSet.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/SVScanDocIdSet.java
index 2633f305d5e..4742c252c53 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/SVScanDocIdSet.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/docidsets/SVScanDocIdSet.java
@@ -38,6 +38,11 @@ public final class SVScanDocIdSet implements BlockDocIdSet {
return _docIdIterator;
}
+ @Override
+ public boolean isScanBased() {
+ return true;
+ }
+
@Override
public long getNumEntriesScannedInFilter() {
return _docIdIterator.getNumEntriesScanned();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator.java
index 3ea299d8a48..a82d59d9ada 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/AndFilterOperator.java
@@ -39,12 +39,19 @@ public class AndFilterOperator extends BaseFilterOperator {
private final List<BaseFilterOperator> _filterOperators;
private final Map<String, String> _queryOptions;
+ private final boolean _restrictionPushdownEnabled;
public AndFilterOperator(List<BaseFilterOperator> filterOperators, @Nullable
Map<String, String> queryOptions,
int numDocs, boolean nullHandlingEnabled) {
+ this(filterOperators, queryOptions, numDocs, nullHandlingEnabled, false);
+ }
+
+ public AndFilterOperator(List<BaseFilterOperator> filterOperators, @Nullable
Map<String, String> queryOptions,
+ int numDocs, boolean nullHandlingEnabled, boolean
restrictionPushdownEnabled) {
super(numDocs, nullHandlingEnabled);
_filterOperators = filterOperators;
_queryOptions = queryOptions;
+ _restrictionPushdownEnabled = restrictionPushdownEnabled;
}
@Override
@@ -66,7 +73,7 @@ public class AndFilterOperator extends BaseFilterOperator {
if (blockDocIdSets.isEmpty()) {
return new MatchAllDocIdSet(_numDocs);
}
- return new AndDocIdSet(blockDocIdSets, _queryOptions);
+ return new AndDocIdSet(blockDocIdSets, _queryOptions,
_restrictionPushdownEnabled);
}
/// A conjunction is not false where no child is false: the intersection of
the children's not-false documents.
@@ -86,7 +93,8 @@ public class AndFilterOperator extends BaseFilterOperator {
if (notFalses.isEmpty()) {
return new MatchAllDocIdSet(_numDocs);
}
- return notFalses.size() == 1 ? notFalses.get(0) : new
AndDocIdSet(notFalses, _queryOptions);
+ return notFalses.size() == 1 ? notFalses.get(0)
+ : new AndDocIdSet(notFalses, _queryOptions,
_restrictionPushdownEnabled);
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/BaseColumnFilterOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/BaseColumnFilterOperator.java
index 16d30ed512a..dc2bc68f879 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/BaseColumnFilterOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/BaseColumnFilterOperator.java
@@ -105,8 +105,11 @@ public abstract class BaseColumnFilterOperator extends
BaseFilterOperator {
BlockDocIdSet nonNulls =
BitmapDocIdSet.create(ImmutableRoaringBitmap.flip(nullBitmap, 0, (long)
_numDocs),
_numDocs);
if (nonNulls instanceof EmptyDocIdSet) {
+ // The match is never evaluated, so a scan-based one will not close its
own iterator
+ blockDocIdSet.release();
return new EmptyDocIdSet(blockDocIdSet.getNumEntriesScannedInFilter());
}
- return new AndDocIdSet(List.of(blockDocIdSet, nonNulls),
_queryContext.getQueryOptions());
+ return new AndDocIdSet(List.of(blockDocIdSet, nonNulls),
_queryContext.getQueryOptions(),
+ _queryContext.isAndRestrictionPushdownEnabled());
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/CombinedFilterOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/CombinedFilterOperator.java
index 52800c141aa..3d1b7f7f588 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/CombinedFilterOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/CombinedFilterOperator.java
@@ -35,9 +35,10 @@ public class CombinedFilterOperator extends
BaseFilterOperator {
private final BaseFilterOperator _mainFilterOperator;
private final BaseFilterOperator _subFilterOperator;
private final Map<String, String> _queryOptions;
+ private final boolean _restrictionPushdownEnabled;
public CombinedFilterOperator(BaseFilterOperator mainFilterOperator,
BaseFilterOperator subFilterOperator,
- Map<String, String> queryOptions) {
+ Map<String, String> queryOptions, boolean restrictionPushdownEnabled) {
// This filter operator does not support AND/OR/NOT operations.
super(0, false);
assert !mainFilterOperator.isResultEmpty() &&
!mainFilterOperator.isResultMatchingAll()
@@ -45,6 +46,7 @@ public class CombinedFilterOperator extends
BaseFilterOperator {
_mainFilterOperator = mainFilterOperator;
_subFilterOperator = subFilterOperator;
_queryOptions = queryOptions;
+ _restrictionPushdownEnabled = restrictionPushdownEnabled;
}
@Override
@@ -65,6 +67,7 @@ public class CombinedFilterOperator extends
BaseFilterOperator {
return EmptyDocIdSet.unscanned();
}
BlockDocIdSet subFilterDocIdSet =
_subFilterOperator.nextBlock().getBlockDocIdSet();
- return new AndDocIdSet(Arrays.asList(mainFilterDocIdSet,
subFilterDocIdSet), _queryOptions);
+ return new AndDocIdSet(Arrays.asList(mainFilterDocIdSet,
subFilterDocIdSet), _queryOptions,
+ _restrictionPushdownEnabled);
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/FilterOperatorUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/FilterOperatorUtils.java
index 29b98784aef..d0757e5b1a9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/FilterOperatorUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/FilterOperatorUtils.java
@@ -175,7 +175,7 @@ public class FilterOperatorUtils {
// Return the AND filter operator with re-ordered child filter
operators
reorderAndFilterChildOperators(queryContext, childFilterOperators);
return new AndFilterOperator(childFilterOperators,
queryContext.getQueryOptions(), numDocs,
- queryContext.isNullHandlingEnabled());
+ queryContext.isNullHandlingEnabled(),
queryContext.isAndRestrictionPushdownEnabled());
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/MapFilterOperator.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/MapFilterOperator.java
index b248684377b..3900a3be3c1 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/MapFilterOperator.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/filter/MapFilterOperator.java
@@ -284,9 +284,10 @@ public class MapFilterOperator extends BaseFilterOperator {
}
/// Forwarded so three-valued logic survives the delegation.
`NotFilterOperator.getTrues()` reads
- /// `getFalses()`, and `And`/`OrFilterOperator.getFalses()` read
`getNulls()`; leaving these on the
- /// base implementation would report an absent key's UNKNOWN docs as FALSE,
so
- /// `NOT (col['key'] = v)` would match every doc missing the key.
+ /// `getFalses()`, and `And`/`OrFilterOperator.getNotFalses()` read their
children's `getNotFalses()`;
+ /// leaving these on the base implementation would report an absent key's
UNKNOWN docs as FALSE, so
+ /// `NOT (col['key'] = v)` would match every doc missing the key.
`getNulls()` is forwarded too, so
+ /// that it stays consistent with the other three.
@Override
protected BlockDocIdSet getNulls() {
return _delegate.getNulls();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
index 3a24e0ed51e..03199b367cb 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/plan/maker/InstancePlanMakerImplV2.java
@@ -56,6 +56,7 @@ import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.utils.CommonConstants.Server;
+import
org.apache.pinot.spi.utils.CommonConstants.Server.AndRestrictionPushdownMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -114,6 +115,8 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
private int _minSegmentGroupTrimSize =
Server.DEFAULT_QUERY_EXECUTOR_MIN_SEGMENT_GROUP_TRIM_SIZE;
private int _minServerGroupTrimSize =
Server.DEFAULT_QUERY_EXECUTOR_MIN_SERVER_GROUP_TRIM_SIZE;
private int _groupByTrimThreshold =
Server.DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD;
+ private AndRestrictionPushdownMode _andRestrictionPushdownMode =
+ Server.DEFAULT_QUERY_EXECUTOR_AND_RESTRICTION_PUSHDOWN_MODE;
@Override
public void init(PinotConfiguration queryExecutorConfig) {
@@ -142,6 +145,9 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
Server.DEFAULT_QUERY_EXECUTOR_MIN_SERVER_GROUP_TRIM_SIZE);
_groupByTrimThreshold =
queryExecutorConfig.getProperty(Server.GROUPBY_TRIM_THRESHOLD,
Server.DEFAULT_QUERY_EXECUTOR_GROUPBY_TRIM_THRESHOLD);
+ _andRestrictionPushdownMode =
QueryOptionsUtils.parseAndRestrictionPushdownMode(
+ Server.AND_RESTRICTION_PUSHDOWN_MODE,
queryExecutorConfig.getProperty(Server.AND_RESTRICTION_PUSHDOWN_MODE,
+
Server.DEFAULT_QUERY_EXECUTOR_AND_RESTRICTION_PUSHDOWN_MODE.name()));
Preconditions.checkState(_groupByTrimThreshold > 0,
"Invalid configurable: groupByTrimThreshold: %d must be positive",
_groupByTrimThreshold);
LOGGER.info("Initialized plan maker with maxExecutionThreads: {},
defaultExecutionThreads: {}, "
@@ -231,6 +237,35 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
new InstanceResponsePlanNode(combinePlanNode, segmentContexts,
fetchContexts, queryContext));
}
+ /// Resolves [AndRestrictionPushdownMode] for a query.
+ ///
+ /// The push-down materializes the filter result instead of streaming it, so
under AUTO it is only applied to a
+ /// query that is going to read every matching document anyway. That is
exactly the aggregation and group-by
+ /// queries:
+ ///
+ /// - a selection query without ORDER BY stops as soon as it has LIMIT rows;
+ /// - a DISTINCT query without ORDER BY stops the same way, in
`DistinctOperator#getNextBlock`;
+ /// - whether a selection query *with* ORDER BY can stop early is decided
per segment, because
+ /// [org.apache.pinot.core.plan.SelectionPlanNode] picks
`SelectionPartiallyOrderedBy*` from the segment's sorted
+ /// column, so it cannot be ruled out here and is excluded conservatively;
+ /// - LIMIT 0 reads nothing at all.
+ ///
+ /// Use ALWAYS to push down on a workload known not to terminate early.
+ private static boolean
isAndRestrictionPushdownEnabled(AndRestrictionPushdownMode mode, QueryContext
queryContext) {
+ switch (mode) {
+ case ALWAYS:
+ return true;
+ case NEVER:
+ return false;
+ case AUTO:
+ // NOTE: DISTINCT is modelled as an aggregation function, so it has to
be excluded explicitly
+ return queryContext.getLimit() > 0 &&
QueryContextUtils.isAggregationQuery(queryContext)
+ && !QueryContextUtils.isDistinctQuery(queryContext);
+ default:
+ throw new IllegalStateException("Unsupported AND restriction push-down
mode: " + mode);
+ }
+ }
+
@VisibleForTesting
void applyQueryOptions(QueryContext queryContext) {
Map<String, String> queryOptions = queryContext.getQueryOptions();
@@ -248,6 +283,12 @@ public class InstancePlanMakerImplV2 implements PlanMaker {
// Set skipScanFilterReorder
queryContext.setSkipScanFilterReorder(QueryOptionsUtils.isSkipScanFilterReorder(queryOptions));
+ // Set andRestrictionPushdown, letting the query option override the
server default
+ AndRestrictionPushdownMode andRestrictionPushdownMode =
+ QueryOptionsUtils.getAndRestrictionPushdownMode(queryOptions);
+
queryContext.setAndRestrictionPushdownEnabled(isAndRestrictionPushdownEnabled(
+ andRestrictionPushdownMode != null ? andRestrictionPushdownMode :
_andRestrictionPushdownMode, queryContext));
+
queryContext.setSkipIndexes(QueryOptionsUtils.getSkipIndexes(queryOptions));
// Set maxExecutionThreads
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
index 4f96018d661..fb59e4dd00e 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
@@ -508,7 +508,8 @@ public class AggregationFunctionUtils {
combinedFilterOperator = mainFilterOperator;
} else {
combinedFilterOperator =
- new CombinedFilterOperator(mainFilterOperator,
subFilterOperator, queryContext.getQueryOptions());
+ new CombinedFilterOperator(mainFilterOperator,
subFilterOperator, queryContext.getQueryOptions(),
+ queryContext.isAndRestrictionPushdownEnabled());
}
List<Pair<Predicate, PredicateEvaluator>> subPredicateEvaluators =
subFilterPlan.getPredicateEvaluators();
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
index 41c12e6adce..83fd2ef62eb 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/request/context/QueryContext.java
@@ -103,6 +103,9 @@ public class QueryContext {
private boolean _skipStarTree;
// Whether to skip reordering scan filters for the query
private boolean _skipScanFilterReorder;
+ // Whether an AND may push the document ids it has matched into its
composite (AND/OR/NOT) children.
+ // Resolved from the query option and the server default by
InstancePlanMakerImplV2#applyQueryOptions.
+ private boolean _andRestrictionPushdownEnabled;
// Maximum number of threads used to execute the query
private int _maxExecutionThreads =
Server.DEFAULT_QUERY_EXECUTOR_MAX_EXECUTION_THREADS;
// The following properties apply to group-by queries
@@ -400,6 +403,17 @@ public class QueryContext {
_skipScanFilterReorder = skipScanFilterReorder;
}
+ /// Whether an AND may hand the document ids matched by its index-based
children to its composite (AND/OR/NOT)
+ /// children, so that a scan-based predicate nested inside an OR is only
evaluated on the documents that can still
+ /// match. See
[org.apache.pinot.spi.utils.CommonConstants.Server.AndRestrictionPushdownMode].
+ public boolean isAndRestrictionPushdownEnabled() {
+ return _andRestrictionPushdownEnabled;
+ }
+
+ public void setAndRestrictionPushdownEnabled(boolean
andRestrictionPushdownEnabled) {
+ _andRestrictionPushdownEnabled = andRestrictionPushdownEnabled;
+ }
+
public int getMaxExecutionThreads() {
return _maxExecutionThreads;
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/docidsets/AndDocIdSetPushdownTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/docidsets/AndDocIdSetPushdownTest.java
new file mode 100644
index 00000000000..21f39678db7
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/docidsets/AndDocIdSetPushdownTest.java
@@ -0,0 +1,465 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.operator.docidsets;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.OptionalInt;
+import java.util.Random;
+import org.apache.pinot.core.common.BlockDocIdIterator;
+import org.apache.pinot.core.common.BlockDocIdSet;
+import
org.apache.pinot.core.operator.dociditerators.RangelessBitmapDocIdIterator;
+import org.apache.pinot.core.operator.dociditerators.ScanBasedDocIdIterator;
+import org.apache.pinot.segment.spi.Constants;
+import org.apache.pinot.spi.utils.Pairs.IntPair;
+import org.roaringbitmap.BatchIterator;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Unit test for the restriction push-down: an AND hands the document ids
matched by its index-based children to its
+/// composite (AND/OR/NOT) children, so that a scan-based predicate nested
inside one of them is only evaluated on the
+/// documents that can still match.
+///
+/// See [issue 19339](https://github.com/apache/pinot/issues/19339).
+public class AndDocIdSetPushdownTest {
+ private static final int NUM_DOCS = 10000;
+ private static final long RANDOM_SEED = System.currentTimeMillis();
+ private static final String ERROR_MESSAGE = "Random seed: " + RANDOM_SEED;
+
+ /// The shape reported in the issue: a scan-based predicate sits in an OR
branch next to an index-based predicate.
+ /// That makes the branch take the eager path in [AndDocIdSet#iterator],
materializing it over the whole segment and
+ /// evaluating the scan on every document its sibling matches, regardless of
how selective the enclosing AND is.
+ @Test
+ public void testScanNestedInsideOrBranchIsRestrictedByEnclosingAnd() {
+ ImmutableRoaringBitmap selective = range(0, 100);
+ ImmutableRoaringBitmap indexedInBranch = range(0, 5000);
+ ImmutableRoaringBitmap otherBranch = range(9000, 9010);
+ ImmutableRoaringBitmap scanMatches = multiplesOf(3);
+
+ // Expected: selective AND ((indexedInBranch AND scanMatches) OR
otherBranch)
+ MutableRoaringBitmap expected = MutableRoaringBitmap.and(indexedInBranch,
scanMatches);
+ expected.or(otherBranch);
+ expected.and(selective);
+
+ CountingScanDocIdSet pushedDownScan = new
CountingScanDocIdSet(scanMatches);
+ BlockDocIdSet withPushdown = orBranchTree(true, pushedDownScan, selective,
indexedInBranch, otherBranch);
+ assertEquals(collectDocIds(withPushdown), expected.toArray());
+ long scannedWithPushdown = pushedDownScan.getNumEntriesScannedInFilter();
+ assertEquals(scannedWithPushdown, selective.getCardinality(),
+ "The scan must only visit the documents matching the enclosing AND");
+ assertEquals(withPushdown.getNumEntriesScannedInFilter(),
scannedWithPushdown,
+ "Entries scanned inside the OR must be reported by the enclosing AND");
+
+ CountingScanDocIdSet unrestrictedScan = new
CountingScanDocIdSet(scanMatches);
+ BlockDocIdSet withoutPushdown =
+ orBranchTree(false, unrestrictedScan, selective, indexedInBranch,
otherBranch);
+ assertEquals(collectDocIds(withoutPushdown), expected.toArray(),
+ "Disabling the push-down must not change the result");
+ long scannedWithoutPushdown =
unrestrictedScan.getNumEntriesScannedInFilter();
+ assertEquals(scannedWithoutPushdown, indexedInBranch.getCardinality(),
+ "Without the push-down the scan visits every document matching its
sibling index predicate");
+ assertTrue(scannedWithPushdown < scannedWithoutPushdown,
+ "The push-down must reduce the number of scanned entries");
+ }
+
+ /// AND(selective, NOT(scan)): the scan only has to be evaluated on the
candidates, because
+ /// `NOT(child) AND candidates == candidates MINUS (child AND candidates)`.
+ @Test
+ public void testScanUnderNotIsRestrictedByEnclosingAnd() {
+ ImmutableRoaringBitmap selective = range(0, 100);
+ ImmutableRoaringBitmap scanMatches = multiplesOf(3);
+
+ MutableRoaringBitmap expected = selective.toMutableRoaringBitmap();
+ expected.andNot(scanMatches);
+
+ CountingScanDocIdSet scan = new CountingScanDocIdSet(scanMatches);
+ BlockDocIdSet docIdSet = new AndDocIdSet(
+ List.of(BitmapDocIdSet.create(selective, NUM_DOCS), new
NotDocIdSet(scan, NUM_DOCS)), null, true);
+
+ assertEquals(collectDocIds(docIdSet), expected.toArray());
+ assertEquals(scan.getNumEntriesScannedInFilter(),
selective.getCardinality(),
+ "The scan under the NOT must only visit the documents matching the
enclosing AND");
+ }
+
+ /// With no index-based child there is nothing to seed the push-down with,
so the AND must fall back to the lazy
+ /// iterator over all of its children, including the deferred composite ones.
+ @Test
+ public void testFallsBackToLazyPathWithoutIndexBasedChild() {
+ ImmutableRoaringBitmap firstBranch = range(0, 50);
+ ImmutableRoaringBitmap secondBranch = range(40, 90);
+ ImmutableRoaringBitmap scanMatches = multiplesOf(3);
+
+ MutableRoaringBitmap expected = firstBranch.toMutableRoaringBitmap();
+ expected.or(secondBranch);
+ expected.and(scanMatches);
+
+ // One scan-based branch keeps the OR deferrable, and without an
index-based child the outer AND has no seed
+ BlockDocIdSet or = new OrDocIdSet(
+ List.of(BitmapDocIdSet.create(firstBranch, NUM_DOCS), new
CountingScanDocIdSet(secondBranch)), NUM_DOCS);
+ assertTrue(or.isApplyAndDeferrable());
+ BlockDocIdSet docIdSet = new AndDocIdSet(List.of(new
CountingScanDocIdSet(scanMatches), or), null, true);
+
+ assertEquals(collectDocIds(docIdSet), expected.toArray());
+ }
+
+ /// The push-down must never change the matching documents, whatever the
shape of the filter tree.
+ @Test
+ public void testRandomTreesMatchTheSameDocumentsWithAndWithoutPushdown() {
+ for (int i = 0; i < 200; i++) {
+ long seed = RANDOM_SEED + i;
+ BlockDocIdSet withPushdown = randomTree(new Random(seed), 3, true);
+ BlockDocIdSet withoutPushdown = randomTree(new Random(seed), 3, false);
+ assertEquals(collectDocIds(withPushdown),
collectDocIds(withoutPushdown), ERROR_MESSAGE + ", iteration: " + i);
+ }
+ }
+
+ /// The candidate set can carry document ids beyond numDocs, because
`BitmapDocIdIterator#getDocIds()` hands out the
+ /// raw bitmap while its `next()` stops at numDocs. An AND cannot introduce
such ids, but a NOT can, so the bound has
+ /// to survive the push-down.
+ @Test
+ public void testNotDoesNotEmitDocumentIdsBeyondNumDocs() {
+ int numDocs = 100;
+ // The candidate bitmap deliberately reaches past numDocs, as a live
mutable-segment bitmap can
+ MutableRoaringBitmap candidates = new MutableRoaringBitmap();
+ candidates.add(0L, 150L);
+ BlockDocIdSet child =
BitmapDocIdSet.create(MutableRoaringBitmap.bitmapOf(1, 2, 3), numDocs);
+
+ ImmutableRoaringBitmap docIds = new NotDocIdSet(child,
numDocs).applyAnd(candidates);
+
+ assertTrue(docIds.isEmpty() || docIds.last() < numDocs,
+ "NOT must not emit document ids at or beyond numDocs, but emitted " +
docIds.last());
+ assertEquals(docIds.getCardinality(), numDocs - 3);
+ // The iterator path is the reference implementation: both must agree
+ assertEquals(docIds.toArray(), collectDocIds(new AndDocIdSet(
+ List.of(BitmapDocIdSet.create(candidates, numDocs), new NotDocIdSet(
+ BitmapDocIdSet.create(MutableRoaringBitmap.bitmapOf(1, 2, 3),
numDocs), numDocs)), null, false)));
+ }
+
+ @Test
+ public void testMatchAllDoesNotEmitDocumentIdsBeyondNumDocs() {
+ int numDocs = 100;
+ MutableRoaringBitmap candidates = new MutableRoaringBitmap();
+ candidates.add(0L, 150L);
+
+ ImmutableRoaringBitmap docIds = new
MatchAllDocIdSet(numDocs).applyAnd(candidates);
+
+ assertEquals(docIds.getCardinality(), numDocs);
+ assertTrue(docIds.last() < numDocs);
+ }
+
+ /// The OR short-circuits once every candidate is matched. The branches it
skips were never evaluated, so nothing
+ /// closes the iterators they built in their constructors.
+ @Test
+ public void testShortCircuitedOrBranchesAreReleased() {
+ ImmutableRoaringBitmap candidates = range(0, 100);
+ // The first branch matches every candidate, so the second is never
evaluated
+ ReleaseTrackingDocIdSet skipped = new ReleaseTrackingDocIdSet(range(0,
100));
+ BlockDocIdSet or =
+ new OrDocIdSet(List.of(BitmapDocIdSet.create(range(0, 100), NUM_DOCS),
skipped), NUM_DOCS);
+
+ or.applyAnd(candidates);
+
+ assertTrue(skipped.isReleased(), "A branch skipped by the short-circuit
must still be released");
+ }
+
+ /// A bitmap branch under an OR is intersected through its raw bitmap, which
can reach past numDocs just like the
+ /// candidate set can. The iterator path drops those ids, so the push-down
has to drop them too.
+ @Test
+ public void testOrDoesNotEmitDocumentIdsBeyondNumDocs() {
+ int numDocs = 100;
+ MutableRoaringBitmap candidates = new MutableRoaringBitmap();
+ candidates.add(0L, 150L);
+
+ ImmutableRoaringBitmap docIds = new OrDocIdSet(
+ List.of(BitmapDocIdSet.create(range(90, 150), numDocs), new
CountingScanDocIdSet(range(0, 10))), numDocs)
+ .applyAnd(candidates);
+
+ assertTrue(docIds.last() < numDocs, "OR must not emit document ids at or
beyond numDocs, but emitted "
+ + docIds.last());
+ // The iterator path is the reference implementation: both must agree
+ assertEquals(docIds.toArray(), collectDocIds(new
AndDocIdSet(List.of(BitmapDocIdSet.create(candidates, numDocs),
+ new OrDocIdSet(List.of(BitmapDocIdSet.create(range(90, 150), numDocs),
new CountingScanDocIdSet(range(0, 10))),
+ numDocs)), null, false)));
+ }
+
+ /// A DocIdSet handed no candidate is never evaluated, so it has to be
released instead.
+ @Test
+ public void testDocIdSetsWithoutCandidatesAreReleased() {
+ MutableRoaringBitmap noCandidates = new MutableRoaringBitmap();
+
+ ReleaseTrackingDocIdSet leaf = new ReleaseTrackingDocIdSet(range(0, 10));
+ leaf.applyAnd(noCandidates);
+ assertTrue(leaf.isReleased(), "A leaf handed no candidate must be
released");
+
+ ReleaseTrackingDocIdSet orBranch = new ReleaseTrackingDocIdSet(range(0,
10));
+ new OrDocIdSet(List.of(orBranch), NUM_DOCS).applyAnd(noCandidates);
+ assertTrue(orBranch.isReleased(), "An OR branch must be released when the
OR is handed no candidate");
+
+ // Every candidate is at or beyond numDocs, so the NOT drops them all
before reaching its child
+ int numDocs = 100;
+ MutableRoaringBitmap candidatesBeyondNumDocs = new MutableRoaringBitmap();
+ candidatesBeyondNumDocs.add(100L, 150L);
+ ReleaseTrackingDocIdSet notChild = new ReleaseTrackingDocIdSet(range(0,
10));
+ new NotDocIdSet(notChild, numDocs).applyAnd(candidatesBeyondNumDocs);
+ assertTrue(notChild.isReleased(), "A NOT child must be released when no
candidate is below numDocs");
+ }
+
+ private static final class ReleaseTrackingDocIdSet implements BlockDocIdSet {
+ private final ImmutableRoaringBitmap _docIds;
+ private boolean _released;
+
+ private ReleaseTrackingDocIdSet(ImmutableRoaringBitmap docIds) {
+ _docIds = docIds;
+ }
+
+ private boolean isReleased() {
+ return _released;
+ }
+
+ @Override
+ public BlockDocIdIterator iterator() {
+ return new RangelessBitmapDocIdIterator(_docIds);
+ }
+
+ @Override
+ public long getNumEntriesScannedInFilter() {
+ return 0L;
+ }
+
+ @Override
+ public void release() {
+ _released = true;
+ }
+ }
+
+ /// Deferring a composite child is only worth it when the subtree still has
a scan to restrict. An index-only OR
+ /// produces the same bitmap either way, so deferring it would only add an
intersection per branch.
+ @Test
+ public void testOnlyScanBearingSubtreesAreDeferred() {
+ BlockDocIdSet indexOnlyOr = new OrDocIdSet(
+ List.of(BitmapDocIdSet.create(range(0, 10), NUM_DOCS),
BitmapDocIdSet.create(range(20, 30), NUM_DOCS)),
+ NUM_DOCS);
+ assertFalse(indexOnlyOr.isApplyAndDeferrable(), "An index-only OR gains
nothing from a candidate set");
+
+ BlockDocIdSet scanBearingOr = new OrDocIdSet(
+ List.of(BitmapDocIdSet.create(range(0, 10), NUM_DOCS), new
CountingScanDocIdSet(range(20, 30))), NUM_DOCS);
+ assertTrue(scanBearingOr.isApplyAndDeferrable());
+
+ // The scan two levels down still counts: that is the shape from the issue
+ BlockDocIdSet nested = new
OrDocIdSet(List.of(BitmapDocIdSet.create(range(0, 10), NUM_DOCS),
+ new AndDocIdSet(List.of(BitmapDocIdSet.create(range(20, 30), NUM_DOCS),
+ new CountingScanDocIdSet(range(20, 30))), null, true)), NUM_DOCS);
+ assertTrue(nested.isApplyAndDeferrable());
+
+ assertFalse(new NotDocIdSet(BitmapDocIdSet.create(range(0, 10), NUM_DOCS),
NUM_DOCS).isApplyAndDeferrable());
+ assertTrue(new NotDocIdSet(new CountingScanDocIdSet(range(0, 10)),
NUM_DOCS).isApplyAndDeferrable());
+ }
+
+ private static BlockDocIdSet orBranchTree(boolean pushdownEnabled,
BlockDocIdSet scan,
+ ImmutableRoaringBitmap selective, ImmutableRoaringBitmap
indexedInBranch, ImmutableRoaringBitmap otherBranch) {
+ BlockDocIdSet branch =
+ new AndDocIdSet(List.of(BitmapDocIdSet.create(indexedInBranch,
NUM_DOCS), scan), null, pushdownEnabled);
+ BlockDocIdSet or = new OrDocIdSet(List.of(branch,
BitmapDocIdSet.create(otherBranch, NUM_DOCS)), NUM_DOCS);
+ return new AndDocIdSet(List.of(BitmapDocIdSet.create(selective, NUM_DOCS),
or), null, pushdownEnabled);
+ }
+
+ /// Builds a random filter tree. Two calls with equally seeded [Random]
instances build the same tree, which is what
+ /// lets the same tree be evaluated with and without the push-down.
+ private static BlockDocIdSet randomTree(Random random, int depth, boolean
pushdownEnabled) {
+ if (depth == 0) {
+ return randomLeaf(random);
+ }
+ switch (random.nextInt(5)) {
+ case 0: {
+ List<BlockDocIdSet> children = randomChildren(random, depth,
pushdownEnabled);
+ return new AndDocIdSet(children, null, pushdownEnabled);
+ }
+ case 1: {
+ List<BlockDocIdSet> children = randomChildren(random, depth,
pushdownEnabled);
+ return new OrDocIdSet(children, NUM_DOCS);
+ }
+ case 2:
+ return new NotDocIdSet(randomTree(random, depth - 1, pushdownEnabled),
NUM_DOCS);
+ default:
+ return randomLeaf(random);
+ }
+ }
+
+ private static List<BlockDocIdSet> randomChildren(Random random, int depth,
boolean pushdownEnabled) {
+ int numChildren = 2 + random.nextInt(2);
+ List<BlockDocIdSet> children = new ArrayList<>(numChildren);
+ for (int i = 0; i < numChildren; i++) {
+ children.add(randomTree(random, depth - 1, pushdownEnabled));
+ }
+ return children;
+ }
+
+ /// Builds a leaf of every kind that [BlockDocIdSet#applyAnd] dispatches on,
so that the bitmap, scan, sorted-range
+ /// and match-all/empty arms are all exercised by the randomized equivalence
test.
+ private static BlockDocIdSet randomLeaf(Random random) {
+ switch (random.nextInt(8)) {
+ case 0:
+ return new MatchAllDocIdSet(NUM_DOCS);
+ case 1:
+ return EmptyDocIdSet.unscanned();
+ case 2:
+ case 3: {
+ // Sorted leaf: inclusive ranges, which applyAnd has to convert to an
exclusive upper bound
+ List<IntPair> docIdRanges = new ArrayList<>();
+ int start = random.nextInt(NUM_DOCS / 2);
+ int numRanges = 1 + random.nextInt(3);
+ for (int i = 0; i < numRanges && start < NUM_DOCS; i++) {
+ int end = Math.min(start + random.nextInt(500), NUM_DOCS - 1);
+ docIdRanges.add(new IntPair(start, end));
+ start = end + 2 + random.nextInt(500);
+ }
+ return SortedDocIdSet.create(docIdRanges);
+ }
+ case 4:
+ case 5:
+ return new CountingScanDocIdSet(randomBitmap(random));
+ default:
+ return BitmapDocIdSet.create(randomBitmap(random), NUM_DOCS);
+ }
+ }
+
+ private static MutableRoaringBitmap randomBitmap(Random random) {
+ MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+ int numRuns = 1 + random.nextInt(8);
+ for (int i = 0; i < numRuns; i++) {
+ int start = random.nextInt(NUM_DOCS);
+ bitmap.add(start, Math.min((long) start + 1 + random.nextInt(2000),
NUM_DOCS));
+ }
+ return bitmap;
+ }
+
+ /// A sorted leaf whose ranges end exactly at the candidate boundaries,
pinning the inclusive-to-exclusive
+ /// conversion in [BlockDocIdSet#applyAnd] against an off-by-one.
+ @Test
+ public void testSortedLeafRangeBoundariesAreInclusive() {
+ BlockDocIdSet sorted = SortedDocIdSet.create(List.of(new IntPair(10, 20),
new IntPair(30, 30)));
+ MutableRoaringBitmap candidates = MutableRoaringBitmap.bitmapOf(9, 10, 20,
21, 29, 30, 31);
+
+ assertEquals(sorted.applyAnd(candidates).toArray(), new int[]{10, 20, 30});
+ }
+
+ private static int[] collectDocIds(BlockDocIdSet docIdSet) {
+ BlockDocIdIterator iterator = docIdSet.iterator();
+ List<Integer> docIds = new ArrayList<>();
+ int docId;
+ while ((docId = iterator.next()) != Constants.EOF) {
+ docIds.add(docId);
+ }
+ return docIds.stream().mapToInt(Integer::intValue).toArray();
+ }
+
+ private static MutableRoaringBitmap range(int start, int endExclusive) {
+ MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+ bitmap.add(start, endExclusive);
+ return bitmap;
+ }
+
+ private static MutableRoaringBitmap multiplesOf(int divisor) {
+ MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+ for (int docId = 0; docId < NUM_DOCS; docId += divisor) {
+ bitmap.add(docId);
+ }
+ return bitmap;
+ }
+
+ /// A scan-based DocIdSet that records how many documents it was asked
about, standing in for a predicate with no
+ /// usable index (a full scan, or `IN_SUBQUERY` evaluated through
`ExpressionScanDocIdIterator`).
+ private static final class CountingScanDocIdSet implements BlockDocIdSet {
+ private final CountingScanDocIdIterator _iterator;
+
+ private CountingScanDocIdSet(ImmutableRoaringBitmap matchingDocIds) {
+ _iterator = new CountingScanDocIdIterator(matchingDocIds);
+ }
+
+ @Override
+ public ScanBasedDocIdIterator iterator() {
+ return _iterator;
+ }
+
+ @Override
+ public boolean isScanBased() {
+ return true;
+ }
+
+ @Override
+ public long getNumEntriesScannedInFilter() {
+ return _iterator.getNumEntriesScanned();
+ }
+ }
+
+ private static final class CountingScanDocIdIterator implements
ScanBasedDocIdIterator {
+ private final ImmutableRoaringBitmap _matchingDocIds;
+ private long _numEntriesScanned;
+ private int _nextDocId;
+
+ private CountingScanDocIdIterator(ImmutableRoaringBitmap matchingDocIds) {
+ _matchingDocIds = matchingDocIds;
+ }
+
+ @Override
+ public MutableRoaringBitmap applyAnd(BatchIterator batchIterator,
OptionalInt firstDoc, OptionalInt lastDoc) {
+ MutableRoaringBitmap docIds = new MutableRoaringBitmap();
+ int[] buffer = new int[OPTIMAL_ITERATOR_BATCH_SIZE];
+ while (batchIterator.hasNext()) {
+ int numDocIds = batchIterator.nextBatch(buffer);
+ for (int i = 0; i < numDocIds; i++) {
+ _numEntriesScanned++;
+ if (_matchingDocIds.contains(buffer[i])) {
+ docIds.add(buffer[i]);
+ }
+ }
+ }
+ return docIds;
+ }
+
+ @Override
+ public int next() {
+ while (_nextDocId < NUM_DOCS) {
+ int docId = _nextDocId++;
+ _numEntriesScanned++;
+ if (_matchingDocIds.contains(docId)) {
+ return docId;
+ }
+ }
+ return Constants.EOF;
+ }
+
+ @Override
+ public int advance(int targetDocId) {
+ _nextDocId = targetDocId;
+ return next();
+ }
+
+ @Override
+ public long getNumEntriesScanned() {
+ return _numEntriesScanned;
+ }
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/AndRestrictionPushdownModeTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/AndRestrictionPushdownModeTest.java
new file mode 100644
index 00000000000..79b1e46ad7e
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/plan/maker/AndRestrictionPushdownModeTest.java
@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.plan.maker;
+
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.pinot.core.query.request.context.QueryContext;
+import
org.apache.pinot.core.query.request.context.utils.QueryContextConverterUtils;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
+import org.apache.pinot.spi.utils.CommonConstants.Server;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests how [InstancePlanMakerImplV2] resolves
[Server.AndRestrictionPushdownMode] into the per-query flag read by
+/// the AND filter operators.
+///
+/// Under `AUTO` the push-down is skipped for a selection query without ORDER
BY: it stops once it has LIMIT rows,
+/// while the push-down materializes the whole filter result. See
+/// [issue 19339](https://github.com/apache/pinot/issues/19339).
+public class AndRestrictionPushdownModeTest {
+ private static final String SELECTION_ONLY = "SELECT col1 FROM testTable
WHERE col2 = 1";
+ private static final String SELECTION_ORDER_BY = "SELECT col1 FROM testTable
WHERE col2 = 1 ORDER BY col1";
+ private static final String AGGREGATION = "SELECT SUM(col1) FROM testTable
WHERE col2 = 1";
+ private static final String GROUP_BY = "SELECT col1, SUM(col3) FROM
testTable WHERE col2 = 1 GROUP BY col1";
+ private static final String DISTINCT = "SELECT DISTINCT col1 FROM testTable
WHERE col2 = 1";
+
+ @Test
+ public void testDefaultIsNever() {
+ assertFalse(resolve(null, AGGREGATION), "The push-down ships disabled");
+ assertFalse(resolve(null, SELECTION_ONLY));
+ }
+
+ @Test
+ public void testAutoSkipsQueriesThatCanStopEarly() {
+ assertFalse(resolve("auto", SELECTION_ONLY), "A selection-only query stops
at its LIMIT");
+ assertFalse(resolve("auto", DISTINCT), "DistinctOperator stops once it has
LIMIT rows");
+ // Whether an ORDER BY selection can stop early is decided per segment, so
AUTO excludes it conservatively
+ assertFalse(resolve("auto", SELECTION_ORDER_BY));
+ assertFalse(resolve("auto", AGGREGATION + " LIMIT 0"), "LIMIT 0 reads
nothing");
+ }
+
+ @Test
+ public void testAutoAppliesToQueriesThatReadEveryMatchingDocument() {
+ assertTrue(resolve("auto", AGGREGATION));
+ assertTrue(resolve("auto", GROUP_BY));
+ }
+
+ @Test
+ public void testAlwaysAppliesEvenToQueriesThatCanStopEarly() {
+ assertTrue(resolve("always", SELECTION_ONLY));
+ assertTrue(resolve("always", DISTINCT));
+ }
+
+ @Test
+ public void testNeverSkipsEvenAggregationQuery() {
+ assertFalse(resolve("never", AGGREGATION));
+ }
+
+ @Test
+ public void testModeIsCaseInsensitive() {
+ assertTrue(resolve("AlWaYs", SELECTION_ONLY));
+ assertTrue(resolve("AuTo", AGGREGATION));
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testInvalidQueryOptionValueIsRejected() {
+ // A boolean is the most likely mistake, since the option name reads like
a toggle
+ resolve("true", AGGREGATION);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testInvalidServerConfigValueIsRejected() {
+ planMakerWithServerDefault("sometimes");
+ }
+
+ @Test
+ public void testServerDefaultAppliesWithoutQueryOption() {
+ InstancePlanMakerImplV2 planMaker = planMakerWithServerDefault("always");
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(SELECTION_ONLY);
+ planMaker.applyQueryOptions(queryContext);
+ assertTrue(queryContext.isAndRestrictionPushdownEnabled(), "The server
default must apply");
+ }
+
+ @Test
+ public void testQueryOptionOverridesServerDefault() {
+ InstancePlanMakerImplV2 planMaker = planMakerWithServerDefault("always");
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(AGGREGATION);
+
queryContext.getQueryOptions().put(QueryOptionKey.AND_RESTRICTION_PUSHDOWN_MODE,
"never");
+ planMaker.applyQueryOptions(queryContext);
+ assertFalse(queryContext.isAndRestrictionPushdownEnabled(), "The query
option must win over the server default");
+ }
+
+ private static InstancePlanMakerImplV2 planMakerWithServerDefault(String
mode) {
+ InstancePlanMakerImplV2 planMaker = new InstancePlanMakerImplV2();
+ planMaker.init(new
PinotConfiguration(Map.of(Server.AND_RESTRICTION_PUSHDOWN_MODE, mode)));
+ return planMaker;
+ }
+
+ /// Resolves the flag for the given query, optionally setting the
`andRestrictionPushdown` query option. With no
+ /// option the server default (NEVER) applies.
+ private static boolean resolve(@Nullable String mode, String query) {
+ QueryContext queryContext =
QueryContextConverterUtils.getQueryContext(query);
+ if (mode != null) {
+
queryContext.getQueryOptions().put(QueryOptionKey.AND_RESTRICTION_PUSHDOWN_MODE,
mode);
+ }
+ new InstancePlanMakerImplV2().applyQueryOptions(queryContext);
+ return queryContext.isAndRestrictionPushdownEnabled();
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/AndRestrictionPushdownNullQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/AndRestrictionPushdownNullQueriesTest.java
new file mode 100644
index 00000000000..f544ef1381f
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/AndRestrictionPushdownNullQueriesTest.java
@@ -0,0 +1,201 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.queries;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.IntPredicate;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.response.broker.BrokerResponseNative;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// End-to-end test of the AND restriction push-down with null handling,
against a segment whose scanned columns hold
+/// both null and non-null values.
+///
+/// `idx` has an inverted index and seeds the push-down; `a` and `b` have
none, so predicates on them are scans. Under
+/// null handling a predicate on a null row is UNKNOWN, and `NOT` keeps
UNKNOWN rows out, so the push-down has to carry
+/// the null documents correctly through `excludeNulls()` and
`getNotFalses()`. Each query is checked against a
+/// row-by-row evaluation of three-valued logic, not only against the
push-down turned off.
+public class AndRestrictionPushdownNullQueriesTest extends BaseQueriesTest {
+ private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(),
"AndRestrictionPushdownNullQueriesTest");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final int NUM_ROWS = 5000;
+ // BaseQueriesTest queries two instances of two copies of the segment
+ private static final int NUM_SEGMENT_COPIES = 4;
+ private static final String[] B_VALUES = {"x", "y", "z"};
+
+ private IndexSegment _indexSegment;
+ private List<IndexSegment> _indexSegments;
+
+ @Override
+ protected String getFilter() {
+ return "";
+ }
+
+ @Override
+ protected IndexSegment getIndexSegment() {
+ return _indexSegment;
+ }
+
+ @Override
+ protected List<IndexSegment> getIndexSegments() {
+ return _indexSegments;
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteDirectory(INDEX_DIR);
+ Schema schema = new Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension("id", DataType.INT)
+ .addSingleValueDimension("idx", DataType.INT)
+ .addSingleValueDimension("a", DataType.INT)
+ .addSingleValueDimension("b", DataType.STRING)
+ .build();
+ TableConfig tableConfig = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setInvertedIndexColumns(List.of("idx"))
+ .setNullHandlingEnabled(true)
+ .build();
+
+ List<GenericRow> rows = new ArrayList<>(NUM_ROWS);
+ for (int i = 0; i < NUM_ROWS; i++) {
+ GenericRow row = new GenericRow();
+ row.putValue("id", i);
+ row.putValue("idx", i % 10);
+ row.putValue("a", a(i));
+ row.putValue("b", b(i));
+ rows.add(row);
+ }
+
+ SegmentGeneratorConfig config = new SegmentGeneratorConfig(tableConfig,
schema);
+ config.setTableName(RAW_TABLE_NAME);
+ config.setSegmentName(SEGMENT_NAME);
+ config.setOutDir(INDEX_DIR.getPath());
+ config.setDefaultNullHandlingEnabled(true);
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(config, new GenericRowRecordReader(rows));
+ driver.build();
+
+ ImmutableSegment segment =
+ ImmutableSegmentLoader.load(new File(INDEX_DIR, SEGMENT_NAME), new
IndexLoadingConfig(tableConfig, schema));
+ _indexSegment = segment;
+ _indexSegments = List.of(segment, segment);
+ }
+
+ @AfterClass
+ public void tearDown() {
+ _indexSegment.destroy();
+ FileUtils.deleteQuietly(INDEX_DIR);
+ }
+
+ private static Integer a(int i) {
+ return i % 7 == 0 ? null : i % 100;
+ }
+
+ private static String b(int i) {
+ return i % 11 == 0 ? null : B_VALUES[i % 3];
+ }
+
+ /// Each filter has `idx = 1` to seed the push-down and an OR/AND over
nullable scans, paired with the rows it must
+ /// match under three-valued logic.
+ @DataProvider
+ public static Object[][] filters() {
+ IntPredicate aTrue = i -> a(i) != null && a(i) > 50;
+ IntPredicate aFalse = i -> a(i) != null && a(i) <= 50;
+ IntPredicate bTrue = i -> b(i) != null && b(i).equals("x");
+ IntPredicate bFalse = i -> b(i) != null && !b(i).equals("x");
+ return new Object[][]{
+ // excludeNulls() on each leaf under the pushed-into OR
+ {"(a > 50 OR b = 'x')", aTrue.or(bTrue)},
+ // NOT(OR) is true where the OR is false: every branch false, none
UNKNOWN
+ {"NOT (a > 50 OR b = 'x')", aFalse.and(bFalse)},
+ // NOT(AND) is true where the AND is false: some branch false;
getNotFalses() builds the AND with the flag
+ {"NOT (a > 50 AND b = 'x')", aFalse.or(bFalse)}
+ };
+ }
+
+ @Test(dataProvider = "filters")
+ public void testPushdownMatchesThreeValuedLogic(String predicate,
IntPredicate expected) {
+ String query = "SELECT COUNT(*), SUM(id) FROM testTable WHERE idx = 1 AND
" + predicate;
+ long expectedCount = 0;
+ long expectedSum = 0;
+ for (int i = 0; i < NUM_ROWS; i++) {
+ if (i % 10 == 1 && expected.test(i)) {
+ expectedCount++;
+ expectedSum += i;
+ }
+ }
+ assertTrue(expectedCount > 0, "The filter must match some rows to be
meaningful");
+
+ BrokerResponseNative disabled =
getBrokerResponse(withNullHandling("never", query));
+ BrokerResponseNative enabled =
getBrokerResponse(withNullHandling("always", query));
+
+ for (BrokerResponseNative response : List.of(disabled, enabled)) {
+ Object[] row = response.getResultTable().getRows().get(0);
+ assertEquals(((Number) row[0]).longValue(), expectedCount *
NUM_SEGMENT_COPIES, predicate);
+ assertEquals(((Number) row[1]).longValue(), expectedSum *
NUM_SEGMENT_COPIES, predicate);
+ }
+ assertTrue(enabled.getNumEntriesScannedInFilter() <
disabled.getNumEntriesScannedInFilter(),
+ "The push-down must reduce the entries scanned in the filter for " +
predicate + ", but scanned "
+ + enabled.getNumEntriesScannedInFilter() + " with it and " +
disabled.getNumEntriesScannedInFilter()
+ + " without");
+ }
+
+ @Test(dataProvider = "filters")
+ public void testPushdownKeepsTheSameRowsForASelectionQuery(String predicate,
IntPredicate expected) {
+ String query = "SELECT id, a, b FROM testTable WHERE idx = 1 AND " +
predicate + " ORDER BY id LIMIT 100000";
+
+ List<Object[]> enabledRows = getBrokerResponse(withNullHandling("always",
query)).getResultTable().getRows();
+ List<Object[]> disabledRows = getBrokerResponse(withNullHandling("never",
query)).getResultTable().getRows();
+
+ assertEquals(enabledRows.size(), disabledRows.size(), predicate);
+ for (int i = 0; i < enabledRows.size(); i++) {
+ assertEquals(enabledRows.get(i), disabledRows.get(i), predicate + ", row
" + i);
+ assertTrue(expected.test((Integer) enabledRows.get(i)[0]), predicate +
", row " + i);
+ }
+ }
+
+ private static String withNullHandling(String mode, String query) {
+ return "SET enableNullHandling = true; SET " +
QueryOptionKey.AND_RESTRICTION_PUSHDOWN_MODE + " = '" + mode
+ + "'; " + query;
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/AndRestrictionPushdownQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/AndRestrictionPushdownQueriesTest.java
new file mode 100644
index 00000000000..c77cdd9e224
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/AndRestrictionPushdownQueriesTest.java
@@ -0,0 +1,109 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.queries;
+
+import java.util.List;
+import org.apache.pinot.common.response.broker.BrokerResponseNative;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// End-to-end test of the AND restriction push-down against a real segment.
+///
+/// The shared `FILTER` is `col1 > x AND col3 BETWEEN y AND z AND col5 =
'gFuH' AND (col6 < w OR col11 NOT IN (..))
+/// AND daysSinceEpoch = d`, i.e. an OR containing scan-based predicates under
an AND with index-based ones -- the
+/// shape from [issue 19339](https://github.com/apache/pinot/issues/19339).
The push-down ships disabled, so this is
+/// the only coverage that runs a real query with it on.
+public class AndRestrictionPushdownQueriesTest extends
BaseSingleValueQueriesTest {
+ private static final String AGGREGATION = "SELECT SUM(column1) FROM
testTable";
+ private static final String SELECTION = "SELECT column1, column5, column11
FROM testTable";
+
+ @Test
+ public void testPushdownScansFewerEntriesAndKeepsTheSameResult() {
+ BrokerResponseNative disabled = getBrokerResponse(withMode("never",
AGGREGATION + FILTER));
+ BrokerResponseNative enabled = getBrokerResponse(withMode("always",
AGGREGATION + FILTER));
+
+ assertRowsEqual(enabled, disabled);
+ assertEquals(enabled.getNumDocsScanned(), disabled.getNumDocsScanned(),
"The same documents must match");
+ assertEquals(enabled.getTotalDocs(), disabled.getTotalDocs());
+ assertTrue(enabled.getNumEntriesScannedInFilter() <
disabled.getNumEntriesScannedInFilter(),
+ "The push-down must reduce the entries scanned in the filter, but
scanned "
+ + enabled.getNumEntriesScannedInFilter() + " with it and " +
disabled.getNumEntriesScannedInFilter()
+ + " without");
+ }
+
+ @Test
+ public void testPushdownKeepsTheSameRowsForASelectionQuery() {
+ // ALWAYS is the only mode that reaches a selection query: AUTO excludes
it because it can stop at its LIMIT
+ assertRowsEqual(getBrokerResponse(withMode("always", SELECTION + FILTER)),
+ getBrokerResponse(withMode("never", SELECTION + FILTER)));
+ }
+
+ @Test
+ public void testAutoLeavesSelectionQueriesAlone() {
+ // A selection query stops at its LIMIT, so with several threads the
number of segments it scans depends on timing.
+ // One thread makes the entries scanned in the filter deterministic.
+ String query = "SET " + QueryOptionKey.MAX_EXECUTION_THREADS + " = 1; " +
SELECTION + FILTER;
+ BrokerResponseNative auto = getBrokerResponse(withMode("auto", query));
+ BrokerResponseNative disabled = getBrokerResponse(withMode("never",
query));
+
+ assertRowsEqual(auto, disabled);
+ assertEquals(auto.getNumEntriesScannedInFilter(),
disabled.getNumEntriesScannedInFilter(),
+ "AUTO must not push down for a selection query");
+ }
+
+ @Test
+ public void testAutoPushesDownForAnAggregationQuery() {
+ assertEquals(getBrokerResponse(withMode("auto", AGGREGATION +
FILTER)).getNumEntriesScannedInFilter(),
+ getBrokerResponse(withMode("always", AGGREGATION +
FILTER)).getNumEntriesScannedInFilter());
+ }
+
+ @Test
+ public void testDefaultLeavesEveryQueryAlone() {
+ assertEquals(getBrokerResponse(AGGREGATION +
FILTER).getNumEntriesScannedInFilter(),
+ getBrokerResponse(withMode("never", AGGREGATION +
FILTER)).getNumEntriesScannedInFilter(),
+ "The push-down must be off by default");
+ }
+
+ /// Filtered aggregations build the outer AND through
CombinedFilterOperator, where both the main and the sub
+ /// filter can be deferrable, so it can reach the lazy fallback arm of
AndDocIdSet#iterator().
+ @Test
+ public void testPushdownForFilteredAggregation() {
+ String query = "SELECT SUM(column1) FILTER (WHERE column3 > 0), COUNT(*)
FROM testTable" + FILTER;
+
+ assertRowsEqual(getBrokerResponse(withMode("always", query)),
getBrokerResponse(withMode("never", query)));
+ }
+
+ private static String withMode(String mode, String query) {
+ return "SET " + QueryOptionKey.AND_RESTRICTION_PUSHDOWN_MODE + " = '" +
mode + "'; " + query;
+ }
+
+ private static void assertRowsEqual(BrokerResponseNative actual,
BrokerResponseNative expected) {
+ List<Object[]> actualRows = actual.getResultTable().getRows();
+ List<Object[]> expectedRows = expected.getResultTable().getRows();
+ assertEquals(actual.getResultTable().getDataSchema(),
expected.getResultTable().getDataSchema());
+ assertEquals(actualRows.size(), expectedRows.size());
+ for (int i = 0; i < actualRows.size(); i++) {
+ assertEquals(actualRows.get(i), expectedRows.get(i), "Row " + i + "
differs");
+ }
+ }
+}
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkAndRestrictionPushdown.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkAndRestrictionPushdown.java
new file mode 100644
index 00000000000..a960426833c
--- /dev/null
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkAndRestrictionPushdown.java
@@ -0,0 +1,208 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.perf;
+
+import java.io.File;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.BlockDocIdIterator;
+import org.apache.pinot.core.common.BlockDocIdSet;
+import org.apache.pinot.core.operator.dociditerators.SVScanDocIdIterator;
+import org.apache.pinot.core.operator.docidsets.AndDocIdSet;
+import org.apache.pinot.core.operator.docidsets.BitmapDocIdSet;
+import org.apache.pinot.core.operator.docidsets.OrDocIdSet;
+import
org.apache.pinot.segment.local.io.writer.impl.FixedBitSVForwardIndexWriter;
+import
org.apache.pinot.segment.local.segment.index.readers.forward.FixedBitSVForwardIndexReaderV2;
+import org.apache.pinot.segment.spi.Constants;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.infra.Blackhole;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+
+
+/// Measures the AND restriction push-down (`andRestrictionPushdownMode`) in
wall-clock time and, with `-prof gc`,
+/// allocation -- not only in `numEntriesScannedInFilter`, which can fall
while both of those rise.
+///
+/// The matrix covers the four dimensions that decide whether the push-down
pays for itself:
+///
+/// - `_shape`: whether the OR subtree still has a scan to restrict
(`SCAN_IN_OR`, the shape from
+/// [issue 19339](https://github.com/apache/pinot/issues/19339)) or is
entirely index-based (`INDEXED_OR`, where
+/// there is nothing to gain and the question is only how much the machinery
costs). `SCAN_ONLY_OR` is the third
+/// case: scans directly under the OR with no index-based sibling, so the OR
is already lazy today and a LIMIT
+/// really does stop early -- the shape the AUTO mode exists to protect.
+/// - `_consume`: whether the consumer drains the filter, as an aggregation
does, or stops at a LIMIT, as a
+/// selection does. The push-down materializes, so this is where it can lose.
+/// - `_candidateSelectivity`: how much the enclosing AND narrows things
before the OR is reached.
+/// - `_pushdown`: the feature itself, off versus on.
+///
+/// Run with `-prof gc` to get `gc.alloc.rate.norm` (bytes per operation)
alongside the timings.
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Fork(1)
+@Warmup(iterations = 3, time = 3)
+@Measurement(iterations = 5, time = 3)
+public class BenchmarkAndRestrictionPushdown {
+ private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(),
"BenchmarkAndRestrictionPushdown");
+ private static final int NUM_DOCS = 2_000_000;
+ private static final int LIMIT = 10;
+ private static final int NUM_DICT_IDS = 1024;
+
+ @Param({"false", "true"})
+ private boolean _pushdown;
+
+ @Param({"SCAN_IN_OR", "SCAN_ONLY_OR", "INDEXED_OR"})
+ private String _shape;
+
+ @Param({"DRAIN", "LIMIT"})
+ private String _consume;
+
+ /// Fraction of documents the enclosing AND's index-based child matches:
sparse candidates versus dense.
+ @Param({"0.01", "0.5"})
+ private double _candidateSelectivity;
+
+ private ImmutableRoaringBitmap _candidateDocIds;
+ private ImmutableRoaringBitmap _firstBranchDocIds;
+ private ImmutableRoaringBitmap _secondBranchDocIds;
+ private PinotDataBuffer _dataBuffer;
+ private FixedBitSVForwardIndexReaderV2 _forwardIndexReader;
+ private BenchmarkScanDocIdIterators.DummyPredicateEvaluator
_predicateEvaluator;
+
+ public static void main(String[] args)
+ throws RunnerException {
+ new Runner(new
OptionsBuilder().include(BenchmarkAndRestrictionPushdown.class.getSimpleName())
+ .addProfiler("gc")
+ .build()).run();
+ }
+
+ @Setup(Level.Trial)
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteDirectory(INDEX_DIR);
+ FileUtils.forceMkdir(INDEX_DIR);
+ File indexFile = new File(INDEX_DIR, "fwd-index");
+ Random random = new Random(42);
+
+ int numBits = 32 - Integer.numberOfLeadingZeros(NUM_DICT_IDS - 1);
+ try (FixedBitSVForwardIndexWriter indexWriter = new
FixedBitSVForwardIndexWriter(indexFile, NUM_DOCS, numBits)) {
+ for (int i = 0; i < NUM_DOCS; i++) {
+ indexWriter.putDictId(random.nextInt(NUM_DICT_IDS));
+ }
+ }
+ _dataBuffer = PinotDataBuffer.mapReadOnlyBigEndianFile(indexFile);
+ _forwardIndexReader = new FixedBitSVForwardIndexReaderV2(_dataBuffer,
NUM_DOCS, numBits);
+ // Matches roughly a third of the documents, so the scan is neither
trivially empty nor a match-all
+ _predicateEvaluator = new
BenchmarkScanDocIdIterators.DummyPredicateEvaluator(NUM_DICT_IDS / 3);
+
+ _candidateDocIds = randomBitmap(random, _candidateSelectivity);
+ _firstBranchDocIds = randomBitmap(random, 0.3);
+ _secondBranchDocIds = randomBitmap(random, 0.2);
+ }
+
+ @TearDown(Level.Trial)
+ public void tearDown()
+ throws Exception {
+ _dataBuffer.close();
+ FileUtils.deleteDirectory(INDEX_DIR);
+ }
+
+ @Benchmark
+ public int evaluateFilter(Blackhole bh) {
+ BlockDocIdIterator docIdIterator = buildFilter().iterator();
+ int limit = "LIMIT".equals(_consume) ? LIMIT : Integer.MAX_VALUE;
+ int numDocs = 0;
+ int docId;
+ while ((docId = docIdIterator.next()) != Constants.EOF) {
+ bh.consume(docId);
+ if (++numDocs >= limit) {
+ break;
+ }
+ }
+ return numDocs;
+ }
+
+ /// `AND(candidates, OR(...))`, rebuilt on every invocation because a
DocIdSet is single-use.
+ private BlockDocIdSet buildFilter() {
+ BlockDocIdSet orDocIdSet;
+ if ("INDEXED_OR".equals(_shape)) {
+ orDocIdSet = new
OrDocIdSet(List.of(BitmapDocIdSet.create(_firstBranchDocIds, NUM_DOCS),
+ BitmapDocIdSet.create(_secondBranchDocIds, NUM_DOCS)), NUM_DOCS);
+ } else if ("SCAN_ONLY_OR".equals(_shape)) {
+ // No index-based sibling inside the branches, so without the push-down
the OR stays lazy and a LIMIT really
+ // does stop early. This is the shape AUTO exists to protect.
+ orDocIdSet = new OrDocIdSet(List.of(newScanDocIdSet(),
newScanDocIdSet()), NUM_DOCS);
+ } else {
+ // The reported shape: a scan sits next to an index-based predicate
inside an OR branch
+ BlockDocIdSet branch = new AndDocIdSet(
+ List.of(BitmapDocIdSet.create(_firstBranchDocIds, NUM_DOCS),
newScanDocIdSet()), null, _pushdown);
+ orDocIdSet =
+ new OrDocIdSet(List.of(branch,
BitmapDocIdSet.create(_secondBranchDocIds, NUM_DOCS)), NUM_DOCS);
+ }
+ return new AndDocIdSet(List.of(BitmapDocIdSet.create(_candidateDocIds,
NUM_DOCS), orDocIdSet), null, _pushdown);
+ }
+
+ private BlockDocIdSet newScanDocIdSet() {
+ SVScanDocIdIterator docIdIterator =
+ new SVScanDocIdIterator(_predicateEvaluator, _forwardIndexReader,
NUM_DOCS);
+ return new BlockDocIdSet() {
+ @Override
+ public BlockDocIdIterator iterator() {
+ return docIdIterator;
+ }
+
+ @Override
+ public boolean isScanBased() {
+ return true;
+ }
+
+ @Override
+ public long getNumEntriesScannedInFilter() {
+ return docIdIterator.getNumEntriesScanned();
+ }
+ };
+ }
+
+ private static ImmutableRoaringBitmap randomBitmap(Random random, double
selectivity) {
+ MutableRoaringBitmap bitmap = new MutableRoaringBitmap();
+ for (int docId = 0; docId < NUM_DOCS; docId++) {
+ if (random.nextDouble() < selectivity) {
+ bitmap.add(docId);
+ }
+ }
+ return bitmap;
+ }
+}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index ee6fd925d58..4b31ac6e346 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -999,6 +999,9 @@ public class CommonConstants {
// Reorder scan based predicates based on cardinality and number of
selected values
public static final String AND_SCAN_REORDERING = "AndScanReordering";
+
+ // Per-query override of Server.AndRestrictionPushdownMode: ALWAYS,
NEVER or AUTO
+ public static final String AND_RESTRICTION_PUSHDOWN_MODE =
"andRestrictionPushdownMode";
public static final String SKIP_INDEXES = "skipIndexes";
// Query option key used to trace rule productions
@@ -1560,6 +1563,28 @@ public class CommonConstants {
QUERY_EXECUTOR_CONFIG_PREFIX +
".heap.usage.throttle.monitorIntervalMs";
public static final long DEFAULT_HEAP_USAGE_THROTTLE_MONITOR_INTERVAL_MS =
1_000L;
+ /// Controls whether an AND pushes the document ids matched by its
index-based children into its composite
+ /// (AND/OR/NOT) children, so that a scan-based predicate nested inside an
OR is only evaluated on the documents
+ /// that can still match the AND. See [issue
19339](https://github.com/apache/pinot/issues/19339).
+ public static final String AND_RESTRICTION_PUSHDOWN_MODE =
"and.restriction.pushdown.mode";
+ public static final String
CONFIG_OF_QUERY_EXECUTOR_AND_RESTRICTION_PUSHDOWN_MODE =
+ QUERY_EXECUTOR_CONFIG_PREFIX + "." + AND_RESTRICTION_PUSHDOWN_MODE;
+ /// Defaults to NEVER: the push-down is a new filter execution path and is
opt-in until it has soaked.
+ public static final AndRestrictionPushdownMode
DEFAULT_QUERY_EXECUTOR_AND_RESTRICTION_PUSHDOWN_MODE =
+ AndRestrictionPushdownMode.NEVER;
+
+ /// How aggressively an AND pushes the document ids it has matched so far
into its composite children.
+ public enum AndRestrictionPushdownMode {
+ /// Always push the restriction down.
+ ALWAYS,
+ /// Never push the restriction down, keeping the behavior Pinot had
before the push-down existed.
+ NEVER,
+ /// Push the restriction down unless the query can stop before reading
every matching document. The push-down
+ /// materializes the filter result, so applying it to a query that stops
at its LIMIT makes it do the full
+ /// filter work instead of only enough to fill the result.
+ AUTO
+ }
+
// Group-by query related configs
public static final String NUM_GROUPS_LIMIT = "num.groups.limit";
public static final String CONFIG_OF_QUERY_EXECUTOR_NUM_GROUPS_LIMIT =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
