You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "Jackie-Jiang (via GitHub)" <gi...@apache.org> on 2026/05/26 21:06:19 UTC
[PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Jackie-Jiang opened a new pull request, #18588:
URL: https://github.com/apache/pinot/pull/18588
## Summary
Refines the two index-based `DISTINCT` operators added in #17872 / #17820 and rewrites their tests to drive the full broker → operator path.
### `JsonIndexDistinctOperator`
- Argument validation moves into the constructor, mirroring `JsonExtractIndexTransformFunction.init`. The operator accepts 3-/4-/5-arg `jsonExtractIndex` calls (path, type, optional default, optional `JSON_MATCH` filter expression) and MV `_ARRAY` types (`INT_ARRAY`, `STRING_ARRAY`, etc.).
- `canUseJsonIndexDistinct` is simplified to a function-name check; the planner routes every `jsonExtractIndex` call through the operator and lets the constructor surface invalid arguments as `IllegalArgumentException`.
- The runtime path intersects per-value doc ids from the JSON index with the `WHERE`-clause filter through a single `remainingDocs` bitmap, so cross-path `JSON_MATCH`, base-column filters, and `IS NULL` filters all use the same code path.
- `numDocsScanned` is populated from the filtered doc set (or total docs when the filter matches everything), so execution statistics line up with the scan/projection path.
- New query option `jsonIndexDistinctSkipMissingPath`: when set, the operator skips parsing the 4-arg default, skips `remainingDocs` tracking, and suppresses the "Illegal Json Path" throw for the 3-arg form. Useful when the caller knows every doc has the path (or doesn't care about misses).
### `InvertedIndexDistinctOperator`
- Caches `_totalDocs` in the constructor instead of recomputing per call.
- DESC-sorted path short-circuits with `intersects` (boolean) rather than `getLongCardinality`, which is orders of magnitude cheaper on dense bitmaps.
- Drops redundant `advanceIfNeeded(startDocId)` on the ASC sorted path and the redundant inner `filterIter.hasNext()` check.
- Reports a correct `numDocsScanned` on the sorted / inverted paths (previously zero).
- Inlines the `FilterPreparation` helper and renames `_numEntriesExamined` → `_numEntriesExaminedPostFilter` so the stats name matches its meaning.
### Tests
- Both operators get a new queries-based test (`JsonIndexDistinctOperatorQueriesTest`, `InvertedIndexDistinctOperatorQueriesTest`) that drives `SELECT DISTINCT` through `BaseQueriesTest`, asserts on result tables, explain strings, and execution statistics (`numDocsScanned`, `numEntriesScannedInFilter`, `numEntriesScannedPostFilter`, `numTotalDocs`).
- The older mock-based unit tests are removed — the queries tests cover the same behaviors against real segments.
- All `OPTION(...)` syntax in the suites is converted to standard `SET a=b;` prefixes; repeated query strings are extracted into shared constants.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3319261324
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/InvertedIndexDistinctOperator.java:
##########
@@ -204,59 +198,15 @@ private boolean shouldUseBitmapInvertedIndex(int filteredDocCount) {
return (double) dictionaryCardinality * costRatio <= filteredDocCount;
}
- static final class FilterPreparation {
- @Nullable
- private final ImmutableRoaringBitmap _filteredDocIds;
- @Nullable
- private final Integer _filteredDocCount;
-
- private FilterPreparation(@Nullable ImmutableRoaringBitmap filteredDocIds, @Nullable Integer filteredDocCount) {
- _filteredDocIds = filteredDocIds;
- _filteredDocCount = filteredDocCount;
- }
-
- @Nullable
- ImmutableRoaringBitmap getFilteredDocIds() {
- return _filteredDocIds;
- }
-
- @Nullable
- Integer getFilteredDocCount() {
- return _filteredDocCount;
- }
- }
-
- FilterPreparation prepareBitmapPathInput() {
- int totalDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
- if (_filterOperator.isResultMatchingAll()) {
- return new FilterPreparation(null, totalDocs);
- }
- if (_filterOperator.isResultEmpty()) {
- return new FilterPreparation(new MutableRoaringBitmap(), 0);
- }
- // Prefer the cheaper exact count when available so scan fallback does not pay eager bitmap materialization.
- if (_filterOperator.canOptimizeCount()) {
Review Comment:
This reduces the repeated bitmap computation on happy path, which should be majority of the cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3319277567
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -589,34 +466,6 @@ protected String getExplainName() {
@Override
protected void explainAttributes(ExplainAttributeBuilder attributeBuilder) {
super.explainAttributes(attributeBuilder);
- List<ExpressionContext> selectExpressions = _queryContext.getSelectExpressions();
- if (!selectExpressions.isEmpty()) {
- attributeBuilder.putStringList("keyColumns",
- List.of(selectExpressions.get(0).toString()));
- }
- }
-
- /**
- * Returns true if the expression is the 3/4-arg scalar jsonExtractIndex form on a column with JSON index and the
- * path is indexed. For OSS JSON index all paths are indexed. For composite JSON index, only paths in
- * invertedIndexConfigs are indexed per key.
- */
- public static boolean canUseJsonIndexDistinct(IndexSegment indexSegment, ExpressionContext expr) {
Review Comment:
Correct. The argument handling is the same in this operator vs scan based operator, so there is no value falling back then fail in the scan based operator
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3319290665
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
Review Comment:
`jsonFilterExpression` should be passed as 5th argument, and the support is added in this PR. This rewrite itself can be fragile, and we are not doing this rewrite for the scan-based operator. If we really need this rewrite, it shouldn't be done here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18588:
URL: https://github.com/apache/pinot/pull/18588#issuecomment-4549344271
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18588?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `52.46637%` with `106 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.78%. Comparing base ([`baccdcc`](https://app.codecov.io/gh/apache/pinot/commit/baccdcc489017df4449f5e594d3fab564b504ab2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`dea8cbd`](https://app.codecov.io/gh/apache/pinot/commit/dea8cbddbebf9b80e00b0636086dd71d6223c4f7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/18588?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...core/operator/query/JsonIndexDistinctOperator.java](https://app.codecov.io/gh/apache/pinot/pull/18588?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Foperator%2Fquery%2FJsonIndexDistinctOperator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9vcGVyYXRvci9xdWVyeS9Kc29uSW5kZXhEaXN0aW5jdE9wZXJhdG9yLmphdmE=) | 44.44% | [74 Missing and 21 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../operator/query/InvertedIndexDistinctOperator.java](https://app.codecov.io/gh/apache/pinot/pull/18588?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Foperator%2Fquery%2FInvertedIndexDistinctOperator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9vcGVyYXRvci9xdWVyeS9JbnZlcnRlZEluZGV4RGlzdGluY3RPcGVyYXRvci5qYXZh) | 81.08% | [2 Missing and 5 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rm/function/JsonExtractIndexTransformFunction.java](https://app.codecov.io/gh/apache/pinot/pull/18588?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Foperator%2Ftransform%2Ffunction%2FJsonExtractIndexTransformFunction.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9vcGVyYXRvci90cmFuc2Zvcm0vZnVuY3Rpb24vSnNvbkV4dHJhY3RJbmRleFRyYW5zZm9ybUZ1bmN0aW9uLmphdmE=) | 66.66% | [2 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (baccdcc) and HEAD (dea8cbd). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (baccdcc) | HEAD (dea8cbd) |
>|------|------|------|
>|java-21|5|1|
>|unittests|2|1|
>|temurin|5|1|
>|unittests2|1|0|
>|integration|3|0|
>|integration1|1|0|
>|integration2|1|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18588 +/- ##
============================================
- Coverage 64.28% 56.78% -7.50%
+ Complexity 1137 1 -1136
============================================
Files 3335 2564 -771
Lines 205898 148841 -57057
Branches 32129 24070 -8059
============================================
- Hits 132355 84519 -47836
+ Misses 62894 57158 -5736
+ Partials 10649 7164 -3485
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.78% <52.46%> (-7.50%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.78% <52.46%> (-7.50%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.78% <52.46%> (-7.50%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.78% <52.46%> (+0.01%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18588/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/18588?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3316431825
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/InvertedIndexDistinctOperator.java:
##########
@@ -204,59 +198,15 @@ private boolean shouldUseBitmapInvertedIndex(int filteredDocCount) {
return (double) dictionaryCardinality * costRatio <= filteredDocCount;
}
- static final class FilterPreparation {
- @Nullable
- private final ImmutableRoaringBitmap _filteredDocIds;
- @Nullable
- private final Integer _filteredDocCount;
-
- private FilterPreparation(@Nullable ImmutableRoaringBitmap filteredDocIds, @Nullable Integer filteredDocCount) {
- _filteredDocIds = filteredDocIds;
- _filteredDocCount = filteredDocCount;
- }
-
- @Nullable
- ImmutableRoaringBitmap getFilteredDocIds() {
- return _filteredDocIds;
- }
-
- @Nullable
- Integer getFilteredDocCount() {
- return _filteredDocCount;
- }
- }
-
- FilterPreparation prepareBitmapPathInput() {
- int totalDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
- if (_filterOperator.isResultMatchingAll()) {
- return new FilterPreparation(null, totalDocs);
- }
- if (_filterOperator.isResultEmpty()) {
- return new FilterPreparation(new MutableRoaringBitmap(), 0);
- }
- // Prefer the cheaper exact count when available so scan fallback does not pay eager bitmap materialization.
- if (_filterOperator.canOptimizeCount()) {
Review Comment:
this code path is removed, will it cause a perf regression?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3319115052
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/InvertedIndexDistinctOperator.java:
##########
@@ -107,16 +104,15 @@ public class InvertedIndexDistinctOperator extends BaseOperator<DistinctResultsB
// Execution tracking
private boolean _usedInvertedIndexPath = false;
Review Comment:
There are minor differences on empty filter cases, but should be fine. Applied the suggested change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3306940982
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
- boolean filterFullyPushedDown = pushedDownFilterJson != null
- && isOnlySamePathJsonMatchFilter(parsed, _queryContext.getFilter())
- && !jsonMatchFilterCanMatchMissingPath(pushedDownFilterJson);
-
- // Fast path: when the filter is fully pushed down into the JSON index, we only need the distinct value strings.
- // This avoids reading posting lists, building per-value bitmaps, and converting flattened doc IDs.
- if (filterFullyPushedDown) {
- Set<String> distinctValues = jsonIndexReader.getMatchingDistinctValues(
- parsed._jsonPathString, pushedDownFilterJson);
- return buildDistinctResultsFromValues(expr, parsed, distinctValues);
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() == ExpressionContext.Type.IDENTIFIER) {
+ DataSource dataSource = indexSegment.getDataSource(firstArgument.getIdentifier());
+ _jsonIndexReader = getJsonIndexReader(dataSource);
+ if (_jsonIndexReader == null) {
+ throw new IllegalStateException("jsonExtractIndex can only be applied on a column with JSON index");
+ }
+ } else {
+ throw new IllegalArgumentException("jsonExtractIndex can only be applied to a raw column");
}
- // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
- RoaringBitmap filteredDocIds = buildFilteredDocIds();
- if (filteredDocIds != null && filteredDocIds.isEmpty()) {
- ColumnDataType earlyColumnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema earlyDataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{earlyColumnDataType});
- OrderByExpressionContext earlyOrderBy = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- return new DistinctResultsBlock(
- createDistinctTable(earlyDataSchema, parsed._dataType, earlyOrderBy), _queryContext);
+ ExpressionContext secondArgument = arguments.get(1);
+ if (secondArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path argument must be a literal");
+ }
+ _jsonPathString = secondArgument.getLiteral().getStringValue();
+ try {
+ JsonPathCache.INSTANCE.getOrCompute(_jsonPathString);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("JSON path argument is not a valid JSON path");
}
Review Comment:
Intentionally dropped — `JsonExtractIndexTransformFunction.getValueToMatchingDocsMap` also calls `getMatchingFlattenedDocsMap` without an `isPathIndexed` gate. For readers with selective path support (e.g. the StarTree composite plugin), that call returns an empty map on non-indexed paths and the operator falls through to `handleMissingDocs`, which throws `Illegal Json Path` for the 3-arg form, adds the literal for the 4-arg form, or is suppressed entirely when `jsonIndexDistinctSkipMissingPath` is set. That matches the transform's per-row contract exactly. Restoring the `isPathIndexed` gate would diverge from the transform and prevent the new query option from suppressing the throw.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
- boolean filterFullyPushedDown = pushedDownFilterJson != null
- && isOnlySamePathJsonMatchFilter(parsed, _queryContext.getFilter())
- && !jsonMatchFilterCanMatchMissingPath(pushedDownFilterJson);
-
- // Fast path: when the filter is fully pushed down into the JSON index, we only need the distinct value strings.
- // This avoids reading posting lists, building per-value bitmaps, and converting flattened doc IDs.
- if (filterFullyPushedDown) {
- Set<String> distinctValues = jsonIndexReader.getMatchingDistinctValues(
- parsed._jsonPathString, pushedDownFilterJson);
- return buildDistinctResultsFromValues(expr, parsed, distinctValues);
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() == ExpressionContext.Type.IDENTIFIER) {
+ DataSource dataSource = indexSegment.getDataSource(firstArgument.getIdentifier());
+ _jsonIndexReader = getJsonIndexReader(dataSource);
+ if (_jsonIndexReader == null) {
+ throw new IllegalStateException("jsonExtractIndex can only be applied on a column with JSON index");
+ }
+ } else {
+ throw new IllegalArgumentException("jsonExtractIndex can only be applied to a raw column");
}
- // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
- RoaringBitmap filteredDocIds = buildFilteredDocIds();
- if (filteredDocIds != null && filteredDocIds.isEmpty()) {
- ColumnDataType earlyColumnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema earlyDataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{earlyColumnDataType});
- OrderByExpressionContext earlyOrderBy = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- return new DistinctResultsBlock(
- createDistinctTable(earlyDataSchema, parsed._dataType, earlyOrderBy), _queryContext);
+ ExpressionContext secondArgument = arguments.get(1);
+ if (secondArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path argument must be a literal");
+ }
+ _jsonPathString = secondArgument.getLiteral().getStringValue();
+ try {
+ JsonPathCache.INSTANCE.getOrCompute(_jsonPathString);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("JSON path argument is not a valid JSON path");
}
- // All other WHERE filters remain row-level and are applied after converting flattened doc IDs to real doc IDs.
- Map<String, RoaringBitmap> valueToMatchingDocs =
- jsonIndexReader.getMatchingFlattenedDocsMap(parsed._jsonPathString, pushedDownFilterJson);
+ ExpressionContext thirdArgument = arguments.get(2);
+ if (thirdArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Result type argument must be a literal");
+ }
+ String resultsType = thirdArgument.getLiteral().getStringValue().toUpperCase();
+ boolean isSingleValue = !resultsType.endsWith("_ARRAY");
+ if (isSingleValue && _jsonPathString.contains("[*]")) {
+ throw new IllegalArgumentException(
+ "[*] syntax in json path is unsupported for singleValue field json_extract_index");
+ }
+ String dataTypeName = isSingleValue ? resultsType : resultsType.substring(0, resultsType.length() - 6);
+ try {
+ _dataType = DataType.valueOf(dataTypeName);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Unknown jsonExtractIndex result type: " + resultsType);
+ }
+ switch (_dataType) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case BIG_DECIMAL:
+ case STRING:
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported jsonExtractIndex result type for distinct: " + _dataType);
+ }
- // Always single-value (MV _ARRAY is rejected in parseJsonExtractIndex)
- jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
- return buildDistinctResultsBlock(expr, parsed, valueToMatchingDocs, filteredDocIds,
- filteredDocIds == null);
- }
+ // With _skipMissingPath, the 4-arg default is never used at runtime (handleMissingDocs is bypassed), so don't
+ // parse or validate it — accept any literal shape and ignore it.
+ if (numArguments >= 4 && !_skipMissingPath) {
+ ExpressionContext fourthArgument = arguments.get(3);
+ if (fourthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Default value must be a literal");
+ }
+ String defaultLiteral = fourthArgument.getLiteral().getStringValue();
+ if (isSingleValue) {
+ try {
+ _dataType.convert(defaultLiteral);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + defaultLiteral + "' is not a valid " + _dataType);
+ }
+ _defaultValueLiterals = new String[]{defaultLiteral};
+ } else {
+ try {
+ JsonNode mvArray = JsonUtils.stringToJsonNode(defaultLiteral);
+ if (!mvArray.isArray()) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ String[] literals = new String[mvArray.size()];
+ for (int i = 0; i < mvArray.size(); i++) {
+ literals[i] = mvArray.get(i).asText();
+ try {
+ _dataType.convert(literals[i]);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + literals[i] + "' is not a valid " + _dataType);
+ }
+ }
+ _defaultValueLiterals = literals;
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ }
+ } else {
+ _defaultValueLiterals = null;
+ }
- private DistinctResultsBlock buildDistinctResultsFromValues(ExpressionContext expr, ParsedJsonExtractIndex parsed,
- Set<String> distinctValues) {
- ColumnDataType columnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema dataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{columnDataType});
- OrderByExpressionContext orderByExpression = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- DistinctTable distinctTable = createDistinctTable(dataSchema, parsed._dataType, orderByExpression);
- int limit = _queryContext.getLimit();
+ if (numArguments == 5) {
+ ExpressionContext fifthArgument = arguments.get(4);
+ if (fifthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path filter argument must be a literal");
+ }
+ _filterJsonExpression = fifthArgument.getLiteral().getStringValue();
+ } else {
+ _filterJsonExpression = null;
+ }
- for (String value : distinctValues) {
- _numEntriesExamined++;
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExamined, EXPLAIN_NAME);
+ _dataSchema = new DataSchema(new String[]{_expression.toString()},
+ new ColumnDataType[]{ColumnDataType.fromDataTypeSV(_dataType)});
+ List<OrderByExpressionContext> orderByExpressions = queryContext.getOrderByExpressions();
+ _orderByExpression = orderByExpressions != null ? orderByExpressions.get(0) : null;
+ }
- boolean done = addValueToDistinctTable(distinctTable, value, parsed._dataType, orderByExpression);
- if (done) {
- break;
- }
- if (orderByExpression == null && distinctTable.hasLimit() && distinctTable.size() >= limit) {
- break;
+ @Nullable
+ private static JsonIndexReader getJsonIndexReader(DataSource dataSource) {
+ JsonIndexReader reader = dataSource.getJsonIndex();
+ // TODO: rework
+ if (reader == null) {
+ Optional<IndexType<?, ?, ?>> compositeIndex = IndexService.getInstance().getOptional("composite_json_index");
+ if (compositeIndex.isPresent()) {
+ reader = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
}
}
+ return reader;
+ }
- return new DistinctResultsBlock(distinctTable, _queryContext);
+ @Override
+ protected DistinctResultsBlock getNextBlock() {
+ // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
+ BaseFilterOperator.FilteredDocIds filteredDocIds = _filterOperator.getFilteredDocIds();
+ ImmutableRoaringBitmap docIds = filteredDocIds.getDocIds();
+ _numDocsScanned = docIds != null ? docIds.getCardinality() : _totalDocs;
+ _numEntriesScannedInFilter = filteredDocIds.getNumEntriesScannedInFilter();
+ if (_numDocsScanned == 0) {
+ return new DistinctResultsBlock(createDistinctTable(), _queryContext);
+ }
+
+ // The 5-arg form's filter literal is pushed into the JSON index; WHERE-clause filters remain row-level and are
+ // applied after converting flattened doc IDs to real doc IDs.
+ Map<String, RoaringBitmap> valueToMatchingDocs =
+ _jsonIndexReader.getMatchingFlattenedDocsMap(_jsonPathString, _filterJsonExpression);
+ _jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
Review Comment:
The transform keeps flattened doc IDs because it has to materialize a per-doc MV array — the per-doc array element grouping matters for its row-by-row output. DISTINCT has no such requirement: it collects the *set* of values across all docs and doesn't care how many times a value appears within one doc.
After `convertFlattenedDocIdsToDocIds`, `"a" → {realDocIds…}` is sufficient because (1) the value is added to the distinct table once regardless of multiplicity, (2) `remainingDocs.andNot(realDocIds)` correctly marks a doc as covered the first time any value maps back to it, and (3) the WHERE-clause filter intersects on real doc IDs on both sides. The result set is identical to what `SELECT DISTINCT mvCol` over the scan path produces (the scan path also flattens MV rows to single values before deduping).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3316470168
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/InvertedIndexDistinctOperator.java:
##########
@@ -107,16 +104,15 @@ public class InvertedIndexDistinctOperator extends BaseOperator<DistinctResultsB
// Execution tracking
private boolean _usedInvertedIndexPath = false;
Review Comment:
`_usedInvertedIndexPath` now is the same as `_projectOperator == null`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3316451613
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
- boolean filterFullyPushedDown = pushedDownFilterJson != null
- && isOnlySamePathJsonMatchFilter(parsed, _queryContext.getFilter())
- && !jsonMatchFilterCanMatchMissingPath(pushedDownFilterJson);
-
- // Fast path: when the filter is fully pushed down into the JSON index, we only need the distinct value strings.
- // This avoids reading posting lists, building per-value bitmaps, and converting flattened doc IDs.
- if (filterFullyPushedDown) {
- Set<String> distinctValues = jsonIndexReader.getMatchingDistinctValues(
- parsed._jsonPathString, pushedDownFilterJson);
- return buildDistinctResultsFromValues(expr, parsed, distinctValues);
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() == ExpressionContext.Type.IDENTIFIER) {
+ DataSource dataSource = indexSegment.getDataSource(firstArgument.getIdentifier());
+ _jsonIndexReader = getJsonIndexReader(dataSource);
+ if (_jsonIndexReader == null) {
+ throw new IllegalStateException("jsonExtractIndex can only be applied on a column with JSON index");
+ }
+ } else {
+ throw new IllegalArgumentException("jsonExtractIndex can only be applied to a raw column");
}
- // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
- RoaringBitmap filteredDocIds = buildFilteredDocIds();
- if (filteredDocIds != null && filteredDocIds.isEmpty()) {
- ColumnDataType earlyColumnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema earlyDataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{earlyColumnDataType});
- OrderByExpressionContext earlyOrderBy = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- return new DistinctResultsBlock(
- createDistinctTable(earlyDataSchema, parsed._dataType, earlyOrderBy), _queryContext);
+ ExpressionContext secondArgument = arguments.get(1);
+ if (secondArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path argument must be a literal");
+ }
+ _jsonPathString = secondArgument.getLiteral().getStringValue();
+ try {
+ JsonPathCache.INSTANCE.getOrCompute(_jsonPathString);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("JSON path argument is not a valid JSON path");
}
- // All other WHERE filters remain row-level and are applied after converting flattened doc IDs to real doc IDs.
- Map<String, RoaringBitmap> valueToMatchingDocs =
- jsonIndexReader.getMatchingFlattenedDocsMap(parsed._jsonPathString, pushedDownFilterJson);
+ ExpressionContext thirdArgument = arguments.get(2);
+ if (thirdArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Result type argument must be a literal");
+ }
+ String resultsType = thirdArgument.getLiteral().getStringValue().toUpperCase();
+ boolean isSingleValue = !resultsType.endsWith("_ARRAY");
+ if (isSingleValue && _jsonPathString.contains("[*]")) {
+ throw new IllegalArgumentException(
+ "[*] syntax in json path is unsupported for singleValue field json_extract_index");
+ }
+ String dataTypeName = isSingleValue ? resultsType : resultsType.substring(0, resultsType.length() - 6);
+ try {
+ _dataType = DataType.valueOf(dataTypeName);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Unknown jsonExtractIndex result type: " + resultsType);
+ }
+ switch (_dataType) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case BIG_DECIMAL:
+ case STRING:
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported jsonExtractIndex result type for distinct: " + _dataType);
+ }
- // Always single-value (MV _ARRAY is rejected in parseJsonExtractIndex)
- jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
- return buildDistinctResultsBlock(expr, parsed, valueToMatchingDocs, filteredDocIds,
- filteredDocIds == null);
- }
+ // With _skipMissingPath, the 4-arg default is never used at runtime (handleMissingDocs is bypassed), so don't
+ // parse or validate it — accept any literal shape and ignore it.
+ if (numArguments >= 4 && !_skipMissingPath) {
+ ExpressionContext fourthArgument = arguments.get(3);
+ if (fourthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Default value must be a literal");
+ }
+ String defaultLiteral = fourthArgument.getLiteral().getStringValue();
+ if (isSingleValue) {
+ try {
+ _dataType.convert(defaultLiteral);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + defaultLiteral + "' is not a valid " + _dataType);
+ }
+ _defaultValueLiterals = new String[]{defaultLiteral};
+ } else {
+ try {
+ JsonNode mvArray = JsonUtils.stringToJsonNode(defaultLiteral);
+ if (!mvArray.isArray()) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ String[] literals = new String[mvArray.size()];
+ for (int i = 0; i < mvArray.size(); i++) {
+ literals[i] = mvArray.get(i).asText();
+ try {
+ _dataType.convert(literals[i]);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + literals[i] + "' is not a valid " + _dataType);
+ }
+ }
+ _defaultValueLiterals = literals;
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ }
+ } else {
+ _defaultValueLiterals = null;
+ }
- private DistinctResultsBlock buildDistinctResultsFromValues(ExpressionContext expr, ParsedJsonExtractIndex parsed,
- Set<String> distinctValues) {
- ColumnDataType columnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema dataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{columnDataType});
- OrderByExpressionContext orderByExpression = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- DistinctTable distinctTable = createDistinctTable(dataSchema, parsed._dataType, orderByExpression);
- int limit = _queryContext.getLimit();
+ if (numArguments == 5) {
+ ExpressionContext fifthArgument = arguments.get(4);
+ if (fifthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path filter argument must be a literal");
+ }
+ _filterJsonExpression = fifthArgument.getLiteral().getStringValue();
+ } else {
+ _filterJsonExpression = null;
+ }
- for (String value : distinctValues) {
- _numEntriesExamined++;
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExamined, EXPLAIN_NAME);
+ _dataSchema = new DataSchema(new String[]{_expression.toString()},
+ new ColumnDataType[]{ColumnDataType.fromDataTypeSV(_dataType)});
+ List<OrderByExpressionContext> orderByExpressions = queryContext.getOrderByExpressions();
+ _orderByExpression = orderByExpressions != null ? orderByExpressions.get(0) : null;
+ }
- boolean done = addValueToDistinctTable(distinctTable, value, parsed._dataType, orderByExpression);
- if (done) {
- break;
- }
- if (orderByExpression == null && distinctTable.hasLimit() && distinctTable.size() >= limit) {
- break;
+ @Nullable
+ private static JsonIndexReader getJsonIndexReader(DataSource dataSource) {
+ JsonIndexReader reader = dataSource.getJsonIndex();
+ // TODO: rework
+ if (reader == null) {
+ Optional<IndexType<?, ?, ?>> compositeIndex = IndexService.getInstance().getOptional("composite_json_index");
+ if (compositeIndex.isPresent()) {
+ reader = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
}
}
+ return reader;
+ }
- return new DistinctResultsBlock(distinctTable, _queryContext);
+ @Override
+ protected DistinctResultsBlock getNextBlock() {
+ // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
+ BaseFilterOperator.FilteredDocIds filteredDocIds = _filterOperator.getFilteredDocIds();
+ ImmutableRoaringBitmap docIds = filteredDocIds.getDocIds();
+ _numDocsScanned = docIds != null ? docIds.getCardinality() : _totalDocs;
+ _numEntriesScannedInFilter = filteredDocIds.getNumEntriesScannedInFilter();
+ if (_numDocsScanned == 0) {
+ return new DistinctResultsBlock(createDistinctTable(), _queryContext);
+ }
+
+ // The 5-arg form's filter literal is pushed into the JSON index; WHERE-clause filters remain row-level and are
+ // applied after converting flattened doc IDs to real doc IDs.
+ Map<String, RoaringBitmap> valueToMatchingDocs =
+ _jsonIndexReader.getMatchingFlattenedDocsMap(_jsonPathString, _filterJsonExpression);
+ _jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
+ return buildDistinctResultsBlock(valueToMatchingDocs, docIds != null ? docIds.toRoaringBitmap() : null);
}
- private DistinctResultsBlock buildDistinctResultsBlock(ExpressionContext expr, ParsedJsonExtractIndex parsed,
- Map<String, RoaringBitmap> valueToMatchingDocs, @Nullable RoaringBitmap filteredDocIds,
- boolean allDocsSelected) {
- ColumnDataType columnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema dataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{columnDataType});
- OrderByExpressionContext orderByExpression = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- DistinctTable distinctTable = createDistinctTable(dataSchema, parsed._dataType, orderByExpression);
+ private DistinctResultsBlock buildDistinctResultsBlock(Map<String, RoaringBitmap> valueToMatchingDocs,
+ @Nullable RoaringBitmap filteredDocIds) {
+ DistinctTable distinctTable = createDistinctTable();
- int limit = _queryContext.getLimit();
- int totalDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
- RoaringBitmap coveredDocs = allDocsSelected ? new RoaringBitmap() : null;
- RoaringBitmap remainingDocs = filteredDocIds != null ? filteredDocIds.clone() : null;
- boolean allDocsCovered = filteredDocIds == null ? !allDocsSelected || totalDocs == 0 : filteredDocIds.isEmpty();
+ // With _skipMissingPath, handleMissingDocs is bypassed — no need to track which docs are still uncovered, so
+ // skip the bitmap allocation and per-iteration `andNot` work entirely.
+ boolean allDocsCovered = _skipMissingPath;
+ RoaringBitmap remainingDocs = _skipMissingPath ? null
+ : (filteredDocIds != null ? filteredDocIds.clone() : RoaringBitmap.bitmapOfRange(0L, _totalDocs));
boolean earlyBreak = false;
for (Map.Entry<String, RoaringBitmap> entry : valueToMatchingDocs.entrySet()) {
- _numEntriesExamined++;
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExamined, EXPLAIN_NAME);
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExaminedPostFilter++, EXPLAIN_NAME);
String value = entry.getKey();
RoaringBitmap docIds = entry.getValue();
- boolean includeValue;
- if (filteredDocIds == null) {
- includeValue = true;
- if (!allDocsCovered && allDocsSelected) {
- coveredDocs.or(docIds);
- if (coveredDocs.getLongCardinality() >= totalDocs) {
- allDocsCovered = true;
- }
- }
- } else {
- includeValue = RoaringBitmap.intersects(docIds, filteredDocIds);
- // Remove matched docs from remaining set in-place (no allocation per value).
- if (!allDocsCovered && includeValue) {
- remainingDocs.andNot(docIds);
- if (remainingDocs.isEmpty()) {
- allDocsCovered = true;
- }
- }
- }
+ // Unfiltered always includes; filtered must intersect the original filter set (not the shrinking
+ // `remainingDocs`, since a value can still belong to the result after all filtered docs are covered).
+ boolean includeValue = filteredDocIds == null || RoaringBitmap.intersects(docIds, filteredDocIds);
- if (includeValue) {
- boolean done = addValueToDistinctTable(distinctTable, value, parsed._dataType, orderByExpression);
- if (done) {
- earlyBreak = true;
- break;
+ if (!allDocsCovered && includeValue) {
+ remainingDocs.andNot(docIds);
+ if (remainingDocs.isEmpty()) {
+ allDocsCovered = true;
}
}
- if (orderByExpression == null && distinctTable.hasLimit() && distinctTable.size() >= limit) {
+ // addValueToDistinctTable returns true exactly when the table has reached its LIMIT (no-ORDER-BY case);
+ // for ORDER-BY or unbounded LIMIT it always returns false. So no separate size check is needed.
+ if (includeValue && addValueToDistinctTable(distinctTable, value)) {
earlyBreak = true;
break;
}
}
if (!earlyBreak && !allDocsCovered) {
- handleMissingDocs(distinctTable, parsed, orderByExpression);
+ handleMissingDocs(distinctTable);
}
return new DistinctResultsBlock(distinctTable, _queryContext);
}
- private void handleMissingDocs(DistinctTable distinctTable, ParsedJsonExtractIndex parsed,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (parsed._defaultValueLiteral != null) {
- addValueToDistinctTable(distinctTable, parsed._defaultValueLiteral, parsed._dataType, orderByExpression);
- } else if (_queryContext.isNullHandlingEnabled()) {
- distinctTable.addNull();
- } else {
- throw new RuntimeException(
- String.format("Illegal Json Path: [%s], for some docIds in segment [%s]",
- parsed._jsonPathString, _indexSegment.getSegmentName()));
- }
- }
-
- @Nullable
- private static String extractSamePathJsonMatchFilter(ParsedJsonExtractIndex parsed, @Nullable FilterContext filter) {
- if (filter == null) {
- return null;
- }
- switch (filter.getType()) {
- case PREDICATE:
- return extractSamePathJsonMatchFilter(parsed, filter.getPredicate());
- case AND:
- String matchingFilter = null;
- for (FilterContext child : filter.getChildren()) {
- String childFilter = extractSamePathJsonMatchFilter(parsed, child);
- if (childFilter == null) {
- continue;
- }
- if (matchingFilter != null) {
- return null;
- }
- matchingFilter = childFilter;
- }
- return matchingFilter;
- default:
- return null;
- }
- }
-
- private static boolean isOnlySamePathJsonMatchFilter(ParsedJsonExtractIndex parsed, @Nullable FilterContext filter) {
- if (filter == null || filter.getType() != FilterContext.Type.PREDICATE) {
- return false;
- }
- return extractSamePathJsonMatchFilter(parsed, filter.getPredicate()) != null;
- }
-
- private static boolean jsonMatchFilterCanMatchMissingPath(String filterJsonString) {
- try {
- FilterContext filter = RequestContextUtils.getFilter(CalciteSqlParser.compileToExpression(filterJsonString));
- return filter.getType() == FilterContext.Type.PREDICATE
- && filter.getPredicate().getType() == Predicate.Type.IS_NULL;
- } catch (Exception e) {
- return false;
- }
- }
-
- @Nullable
- private static String extractSamePathJsonMatchFilter(ParsedJsonExtractIndex parsed, Predicate predicate) {
- if (!(predicate instanceof JsonMatchPredicate)) {
- return null;
- }
- ExpressionContext lhs = predicate.getLhs();
- if (lhs.getType() != ExpressionContext.Type.IDENTIFIER
- || !parsed._columnName.equals(lhs.getIdentifier())) {
- return null;
- }
- String filterJsonString = ((JsonMatchPredicate) predicate).getValue();
- int start = filterJsonString.indexOf('"');
- if (start < 0) {
- return null;
- }
- int end = filterJsonString.indexOf('"', start + 1);
- if (end < 0) {
- return null;
- }
- String filterPath = filterJsonString.substring(start + 1, end);
- return parsed._jsonPathString.equals(filterPath) ? filterJsonString : null;
- }
-
- private DistinctTable createDistinctTable(DataSchema dataSchema, FieldSpec.DataType dataType,
- @Nullable OrderByExpressionContext orderByExpression) {
+ private DistinctTable createDistinctTable() {
int limit = _queryContext.getLimit();
boolean nullHandlingEnabled = _queryContext.isNullHandlingEnabled();
- switch (dataType) {
+ switch (_dataType) {
case INT:
- return new IntDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new IntDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case LONG:
- return new LongDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new LongDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case FLOAT:
- return new FloatDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new FloatDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case DOUBLE:
- return new DoubleDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new DoubleDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case BIG_DECIMAL:
- return new BigDecimalDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new BigDecimalDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case STRING:
- return new StringDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new StringDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
default:
- throw new IllegalStateException("Unsupported data type for JSON index distinct: " + dataType);
+ throw new IllegalStateException("Unsupported data type for JSON index distinct: " + _dataType);
}
}
- private static boolean addValueToDistinctTable(DistinctTable distinctTable, String stringValue,
- FieldSpec.DataType dataType, @Nullable OrderByExpressionContext orderByExpression) {
- switch (dataType) {
+ private boolean addValueToDistinctTable(DistinctTable distinctTable, String stringValue) {
+ switch (_dataType) {
case INT:
- return addToTable((IntDistinctTable) distinctTable, Integer.parseInt(stringValue), orderByExpression);
+ return addToTable((IntDistinctTable) distinctTable, Integer.parseInt(stringValue));
case LONG:
- return addToTable((LongDistinctTable) distinctTable, Long.parseLong(stringValue), orderByExpression);
+ return addToTable((LongDistinctTable) distinctTable, Long.parseLong(stringValue));
case FLOAT:
- return addToTable((FloatDistinctTable) distinctTable, Float.parseFloat(stringValue), orderByExpression);
+ return addToTable((FloatDistinctTable) distinctTable, Float.parseFloat(stringValue));
case DOUBLE:
- return addToTable((DoubleDistinctTable) distinctTable, Double.parseDouble(stringValue), orderByExpression);
+ return addToTable((DoubleDistinctTable) distinctTable, Double.parseDouble(stringValue));
case BIG_DECIMAL:
- return addToTable((BigDecimalDistinctTable) distinctTable, new BigDecimal(stringValue), orderByExpression);
+ return addToTable((BigDecimalDistinctTable) distinctTable, new BigDecimal(stringValue));
case STRING:
- return addToTable((StringDistinctTable) distinctTable, stringValue, orderByExpression);
+ return addToTable((StringDistinctTable) distinctTable, stringValue);
default:
- throw new IllegalStateException("Unsupported data type for JSON index distinct: " + dataType);
+ throw new IllegalStateException("Unsupported data type for JSON index distinct: " + _dataType);
}
}
- private static boolean addToTable(IntDistinctTable table, int value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(IntDistinctTable table, int value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
- }
-
- private static boolean addToTable(LongDistinctTable table, long value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
- table.addUnbounded(value);
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
return false;
}
+ return table.addWithoutOrderBy(value);
}
- private static boolean addToTable(FloatDistinctTable table, float value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(LongDistinctTable table, long value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
- }
-
- private static boolean addToTable(DoubleDistinctTable table, double value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
- table.addUnbounded(value);
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
return false;
}
+ return table.addWithoutOrderBy(value);
}
- private static boolean addToTable(BigDecimalDistinctTable table, BigDecimal value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(FloatDistinctTable table, float value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
+ }
+ return table.addWithoutOrderBy(value);
}
- private static boolean addToTable(StringDistinctTable table, String value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(DoubleDistinctTable table, double value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
- }
-
- @Nullable
- private static JsonIndexReader getJsonIndexReader(DataSource dataSource) {
- JsonIndexReader reader = dataSource.getJsonIndex();
- if (reader == null) {
- Optional<IndexType<?, ?, ?>> compositeIndex =
- IndexService.getInstance().getOptional("composite_json_index");
- if (compositeIndex.isPresent()) {
- reader = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
- }
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
}
- return reader;
- }
-
- @Nullable
- private RoaringBitmap buildFilteredDocIds() {
- BaseFilterOperator.FilteredDocIds filteredDocIds = _filterOperator.getFilteredDocIds();
- _numEntriesScannedInFilter = filteredDocIds.getNumEntriesScannedInFilter();
- ImmutableRoaringBitmap docIds = filteredDocIds.getDocIds();
- return docIds != null ? docIds.toRoaringBitmap() : null;
+ return table.addWithoutOrderBy(value);
}
- @Nullable
- private static ParsedJsonExtractIndex parseJsonExtractIndex(ExpressionContext expr) {
- if (expr.getType() != ExpressionContext.Type.FUNCTION) {
- return null;
- }
- if (!FUNCTION_NAME.equalsIgnoreCase(expr.getFunction().getFunctionName())) {
- return null;
- }
- List<ExpressionContext> args = expr.getFunction().getArguments();
- if (args.size() != 3 && args.size() != 4) {
- return null;
- }
- if (args.get(0).getType() != ExpressionContext.Type.IDENTIFIER) {
- return null;
- }
- if (args.get(1).getType() != ExpressionContext.Type.LITERAL
- || args.get(2).getType() != ExpressionContext.Type.LITERAL
- || (args.size() == 4 && args.get(3).getType() != ExpressionContext.Type.LITERAL)) {
- return null;
- }
-
- String columnName = args.get(0).getIdentifier();
- String jsonPathString = args.get(1).getLiteral().getStringValue();
- String resultsType = args.get(2).getLiteral().getStringValue().toUpperCase();
- // Only single-value types are supported; MV (_ARRAY) would have incorrect flattened-to-real
- // docId intersection since convertFlattenedDocIdsToDocIds is skipped for MV.
- if (resultsType.endsWith("_ARRAY")) {
- return null;
- }
- if (jsonPathString.contains("[*]")) {
- return null;
- }
-
- FieldSpec.DataType dataType;
- try {
- dataType = FieldSpec.DataType.valueOf(resultsType);
- } catch (IllegalArgumentException e) {
- return null;
+ private boolean addToTable(BigDecimalDistinctTable table, BigDecimal value) {
+ if (!table.hasLimit()) {
+ table.addUnbounded(value);
+ return false;
}
- // Only types with a corresponding DistinctTable implementation are supported
- switch (dataType) {
- case INT:
- case LONG:
- case FLOAT:
- case DOUBLE:
- case BIG_DECIMAL:
- case STRING:
- break;
- default:
- return null;
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
}
+ return table.addWithoutOrderBy(value);
+ }
- try {
- JsonPathCache.INSTANCE.getOrCompute(jsonPathString);
- } catch (Exception e) {
- return null;
+ private boolean addToTable(StringDistinctTable table, String value) {
+ if (!table.hasLimit()) {
+ table.addUnbounded(value);
+ return false;
}
-
- String defaultValueLiteral = null;
- if (args.size() == 4) {
- defaultValueLiteral = args.get(3).getLiteral().getStringValue();
- try {
- dataType.convert(defaultValueLiteral);
- } catch (Exception e) {
- return null;
- }
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
}
-
- return new ParsedJsonExtractIndex(columnName, jsonPathString, dataType, defaultValueLiteral);
+ return table.addWithoutOrderBy(value);
}
- private static final class ParsedJsonExtractIndex {
- final String _columnName;
- final String _jsonPathString;
- final FieldSpec.DataType _dataType;
- @Nullable
- final String _defaultValueLiteral;
-
- ParsedJsonExtractIndex(String columnName, String jsonPathString, FieldSpec.DataType dataType,
- @Nullable String defaultValueLiteral) {
- _columnName = columnName;
- _jsonPathString = jsonPathString;
- _dataType = dataType;
- _defaultValueLiteral = defaultValueLiteral;
+ private void handleMissingDocs(DistinctTable distinctTable) {
+ if (_defaultValueLiterals != null) {
+ for (String literal : _defaultValueLiterals) {
+ if (addValueToDistinctTable(distinctTable, literal)) {
+ return;
+ }
+ }
+ } else if (_queryContext.isNullHandlingEnabled()) {
+ distinctTable.addNull();
+ } else {
+ throw new RuntimeException(
+ String.format("Illegal Json Path: [%s], for some docIds in segment [%s]", _jsonPathString,
Review Comment:
this makes user error surfaced as internal error?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3316470168
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/InvertedIndexDistinctOperator.java:
##########
@@ -107,16 +104,15 @@ public class InvertedIndexDistinctOperator extends BaseOperator<DistinctResultsB
// Execution tracking
private boolean _usedInvertedIndexPath = false;
Review Comment:
`_usedInvertedIndexPath` now is redundant and the same as `_projectOperator == null`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18588:
URL: https://github.com/apache/pinot/pull/18588#issuecomment-4587269701
Opened the matching docs PR for this change: https://github.com/pinot-contrib/pinot-docs/pull/833
It documents `jsonIndexDistinctSkipMissingPath` and updates the JSON index DISTINCT guidance to match the merged behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3316419347
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -589,34 +466,6 @@ protected String getExplainName() {
@Override
protected void explainAttributes(ExplainAttributeBuilder attributeBuilder) {
super.explainAttributes(attributeBuilder);
- List<ExpressionContext> selectExpressions = _queryContext.getSelectExpressions();
- if (!selectExpressions.isEmpty()) {
- attributeBuilder.putStringList("keyColumns",
- List.of(selectExpressions.get(0).toString()));
- }
- }
-
- /**
- * Returns true if the expression is the 3/4-arg scalar jsonExtractIndex form on a column with JSON index and the
- * path is indexed. For OSS JSON index all paths are indexed. For composite JSON index, only paths in
- * invertedIndexConfigs are indexed per key.
- */
- public static boolean canUseJsonIndexDistinct(IndexSegment indexSegment, ExpressionContext expr) {
Review Comment:
Regression: the previous working query which fallback to default scan-based operator will now fail.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -589,34 +466,6 @@ protected String getExplainName() {
@Override
protected void explainAttributes(ExplainAttributeBuilder attributeBuilder) {
super.explainAttributes(attributeBuilder);
- List<ExpressionContext> selectExpressions = _queryContext.getSelectExpressions();
- if (!selectExpressions.isEmpty()) {
- attributeBuilder.putStringList("keyColumns",
- List.of(selectExpressions.get(0).toString()));
- }
- }
-
- /**
- * Returns true if the expression is the 3/4-arg scalar jsonExtractIndex form on a column with JSON index and the
- * path is indexed. For OSS JSON index all paths are indexed. For composite JSON index, only paths in
- * invertedIndexConfigs are indexed per key.
- */
- public static boolean canUseJsonIndexDistinct(IndexSegment indexSegment, ExpressionContext expr) {
Review Comment:
Regression: the previous working query which fallback to default scan-based operator will now fail?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3316442417
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
Review Comment:
Would removing this short circuit cause perf regression?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3306883749
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
- boolean filterFullyPushedDown = pushedDownFilterJson != null
- && isOnlySamePathJsonMatchFilter(parsed, _queryContext.getFilter())
- && !jsonMatchFilterCanMatchMissingPath(pushedDownFilterJson);
-
- // Fast path: when the filter is fully pushed down into the JSON index, we only need the distinct value strings.
- // This avoids reading posting lists, building per-value bitmaps, and converting flattened doc IDs.
- if (filterFullyPushedDown) {
- Set<String> distinctValues = jsonIndexReader.getMatchingDistinctValues(
- parsed._jsonPathString, pushedDownFilterJson);
- return buildDistinctResultsFromValues(expr, parsed, distinctValues);
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() == ExpressionContext.Type.IDENTIFIER) {
+ DataSource dataSource = indexSegment.getDataSource(firstArgument.getIdentifier());
+ _jsonIndexReader = getJsonIndexReader(dataSource);
+ if (_jsonIndexReader == null) {
+ throw new IllegalStateException("jsonExtractIndex can only be applied on a column with JSON index");
+ }
+ } else {
+ throw new IllegalArgumentException("jsonExtractIndex can only be applied to a raw column");
}
- // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
- RoaringBitmap filteredDocIds = buildFilteredDocIds();
- if (filteredDocIds != null && filteredDocIds.isEmpty()) {
- ColumnDataType earlyColumnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema earlyDataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{earlyColumnDataType});
- OrderByExpressionContext earlyOrderBy = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- return new DistinctResultsBlock(
- createDistinctTable(earlyDataSchema, parsed._dataType, earlyOrderBy), _queryContext);
+ ExpressionContext secondArgument = arguments.get(1);
+ if (secondArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path argument must be a literal");
+ }
+ _jsonPathString = secondArgument.getLiteral().getStringValue();
+ try {
+ JsonPathCache.INSTANCE.getOrCompute(_jsonPathString);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("JSON path argument is not a valid JSON path");
}
Review Comment:
This path no longer checks `_jsonIndexReader.isPathIndexed(_jsonPathString)` after moving `canUseJsonIndexDistinct` to a function-name-only gate. For JSON index readers with selective path support, `DistinctPlanNode` now routes non-indexed paths into this operator instead of falling back, so the operator can call `getMatchingFlattenedDocsMap` for a path the reader explicitly reports as unsupported. Please restore the indexed-path check here (or make the planner fall back when it is false).
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
- boolean filterFullyPushedDown = pushedDownFilterJson != null
- && isOnlySamePathJsonMatchFilter(parsed, _queryContext.getFilter())
- && !jsonMatchFilterCanMatchMissingPath(pushedDownFilterJson);
-
- // Fast path: when the filter is fully pushed down into the JSON index, we only need the distinct value strings.
- // This avoids reading posting lists, building per-value bitmaps, and converting flattened doc IDs.
- if (filterFullyPushedDown) {
- Set<String> distinctValues = jsonIndexReader.getMatchingDistinctValues(
- parsed._jsonPathString, pushedDownFilterJson);
- return buildDistinctResultsFromValues(expr, parsed, distinctValues);
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() == ExpressionContext.Type.IDENTIFIER) {
+ DataSource dataSource = indexSegment.getDataSource(firstArgument.getIdentifier());
+ _jsonIndexReader = getJsonIndexReader(dataSource);
+ if (_jsonIndexReader == null) {
+ throw new IllegalStateException("jsonExtractIndex can only be applied on a column with JSON index");
+ }
+ } else {
+ throw new IllegalArgumentException("jsonExtractIndex can only be applied to a raw column");
}
- // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
- RoaringBitmap filteredDocIds = buildFilteredDocIds();
- if (filteredDocIds != null && filteredDocIds.isEmpty()) {
- ColumnDataType earlyColumnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema earlyDataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{earlyColumnDataType});
- OrderByExpressionContext earlyOrderBy = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- return new DistinctResultsBlock(
- createDistinctTable(earlyDataSchema, parsed._dataType, earlyOrderBy), _queryContext);
+ ExpressionContext secondArgument = arguments.get(1);
+ if (secondArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path argument must be a literal");
+ }
+ _jsonPathString = secondArgument.getLiteral().getStringValue();
+ try {
+ JsonPathCache.INSTANCE.getOrCompute(_jsonPathString);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("JSON path argument is not a valid JSON path");
}
- // All other WHERE filters remain row-level and are applied after converting flattened doc IDs to real doc IDs.
- Map<String, RoaringBitmap> valueToMatchingDocs =
- jsonIndexReader.getMatchingFlattenedDocsMap(parsed._jsonPathString, pushedDownFilterJson);
+ ExpressionContext thirdArgument = arguments.get(2);
+ if (thirdArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Result type argument must be a literal");
+ }
+ String resultsType = thirdArgument.getLiteral().getStringValue().toUpperCase();
+ boolean isSingleValue = !resultsType.endsWith("_ARRAY");
+ if (isSingleValue && _jsonPathString.contains("[*]")) {
+ throw new IllegalArgumentException(
+ "[*] syntax in json path is unsupported for singleValue field json_extract_index");
+ }
+ String dataTypeName = isSingleValue ? resultsType : resultsType.substring(0, resultsType.length() - 6);
+ try {
+ _dataType = DataType.valueOf(dataTypeName);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Unknown jsonExtractIndex result type: " + resultsType);
+ }
+ switch (_dataType) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case BIG_DECIMAL:
+ case STRING:
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported jsonExtractIndex result type for distinct: " + _dataType);
+ }
- // Always single-value (MV _ARRAY is rejected in parseJsonExtractIndex)
- jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
- return buildDistinctResultsBlock(expr, parsed, valueToMatchingDocs, filteredDocIds,
- filteredDocIds == null);
- }
+ // With _skipMissingPath, the 4-arg default is never used at runtime (handleMissingDocs is bypassed), so don't
+ // parse or validate it — accept any literal shape and ignore it.
+ if (numArguments >= 4 && !_skipMissingPath) {
+ ExpressionContext fourthArgument = arguments.get(3);
+ if (fourthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Default value must be a literal");
+ }
+ String defaultLiteral = fourthArgument.getLiteral().getStringValue();
+ if (isSingleValue) {
+ try {
+ _dataType.convert(defaultLiteral);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + defaultLiteral + "' is not a valid " + _dataType);
+ }
+ _defaultValueLiterals = new String[]{defaultLiteral};
+ } else {
+ try {
+ JsonNode mvArray = JsonUtils.stringToJsonNode(defaultLiteral);
+ if (!mvArray.isArray()) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ String[] literals = new String[mvArray.size()];
+ for (int i = 0; i < mvArray.size(); i++) {
+ literals[i] = mvArray.get(i).asText();
+ try {
+ _dataType.convert(literals[i]);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + literals[i] + "' is not a valid " + _dataType);
+ }
+ }
+ _defaultValueLiterals = literals;
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ }
+ } else {
+ _defaultValueLiterals = null;
+ }
- private DistinctResultsBlock buildDistinctResultsFromValues(ExpressionContext expr, ParsedJsonExtractIndex parsed,
- Set<String> distinctValues) {
- ColumnDataType columnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema dataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{columnDataType});
- OrderByExpressionContext orderByExpression = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- DistinctTable distinctTable = createDistinctTable(dataSchema, parsed._dataType, orderByExpression);
- int limit = _queryContext.getLimit();
+ if (numArguments == 5) {
+ ExpressionContext fifthArgument = arguments.get(4);
+ if (fifthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path filter argument must be a literal");
+ }
+ _filterJsonExpression = fifthArgument.getLiteral().getStringValue();
+ } else {
+ _filterJsonExpression = null;
+ }
- for (String value : distinctValues) {
- _numEntriesExamined++;
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExamined, EXPLAIN_NAME);
+ _dataSchema = new DataSchema(new String[]{_expression.toString()},
+ new ColumnDataType[]{ColumnDataType.fromDataTypeSV(_dataType)});
+ List<OrderByExpressionContext> orderByExpressions = queryContext.getOrderByExpressions();
+ _orderByExpression = orderByExpressions != null ? orderByExpressions.get(0) : null;
+ }
- boolean done = addValueToDistinctTable(distinctTable, value, parsed._dataType, orderByExpression);
- if (done) {
- break;
- }
- if (orderByExpression == null && distinctTable.hasLimit() && distinctTable.size() >= limit) {
- break;
+ @Nullable
+ private static JsonIndexReader getJsonIndexReader(DataSource dataSource) {
+ JsonIndexReader reader = dataSource.getJsonIndex();
+ // TODO: rework
+ if (reader == null) {
+ Optional<IndexType<?, ?, ?>> compositeIndex = IndexService.getInstance().getOptional("composite_json_index");
+ if (compositeIndex.isPresent()) {
+ reader = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
}
}
+ return reader;
+ }
- return new DistinctResultsBlock(distinctTable, _queryContext);
+ @Override
+ protected DistinctResultsBlock getNextBlock() {
+ // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
+ BaseFilterOperator.FilteredDocIds filteredDocIds = _filterOperator.getFilteredDocIds();
+ ImmutableRoaringBitmap docIds = filteredDocIds.getDocIds();
+ _numDocsScanned = docIds != null ? docIds.getCardinality() : _totalDocs;
+ _numEntriesScannedInFilter = filteredDocIds.getNumEntriesScannedInFilter();
+ if (_numDocsScanned == 0) {
+ return new DistinctResultsBlock(createDistinctTable(), _queryContext);
+ }
+
+ // The 5-arg form's filter literal is pushed into the JSON index; WHERE-clause filters remain row-level and are
+ // applied after converting flattened doc IDs to real doc IDs.
+ Map<String, RoaringBitmap> valueToMatchingDocs =
+ _jsonIndexReader.getMatchingFlattenedDocsMap(_jsonPathString, _filterJsonExpression);
+ _jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
Review Comment:
The constructor now accepts `_ARRAY` result types, but this always converts the value posting lists from flattened doc IDs to real doc IDs. `JsonExtractIndexTransformFunction` deliberately skips that conversion for MV results because MV extraction needs the flattened posting lists; after conversion, a value that occurs multiple times in one document collapses to a single real doc id and intersections/default handling no longer match the scan path. Gate this conversion on the single-value case or keep rejecting `_ARRAY` types until the MV path can preserve flattened IDs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #18588:
URL: https://github.com/apache/pinot/pull/18588#discussion_r3319300463
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/query/JsonIndexDistinctOperator.java:
##########
@@ -50,516 +47,396 @@
import org.apache.pinot.core.query.distinct.table.StringDistinctTable;
import org.apache.pinot.core.query.request.context.QueryContext;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.segment.spi.SegmentContext;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.IndexService;
import org.apache.pinot.segment.spi.index.IndexType;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.query.QueryThreadContext;
-import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
-/**
- * Distinct operator for the scalar {@code jsonExtractIndex(column, path, type[, defaultValue])} form.
- *
- * <p>Execution flow:
- * 1. Push a same-path {@code JSON_MATCH} predicate into the JSON-index lookup when it cannot match missing paths.
- * 2. Convert matching flattened doc ids back to segment doc ids.
- * 3. Apply any remaining row-level filter and materialize DISTINCT results, including missing-path handling.
- */
+/// Distinct operator for `jsonExtractIndex(column, path, type[, defaultValue[, filterJsonExpression]])`.
+///
+/// Supports both SV (e.g. `STRING`) and MV (e.g. `STRING_ARRAY`) result types — DISTINCT collapses MV array elements
+/// to scalar rows, matching the scan-based `SELECT DISTINCT mvCol` convention. The 4-arg default is a single value
+/// for SV; for MV it's a JSON array whose elements are each added to the distinct set when no doc matches the path.
+///
+/// Execution flow:
+/// 1. Pass the optional 5-arg `filterJsonExpression` directly to the JSON-index lookup (matches
+/// `JsonExtractIndexTransformFunction`'s convention).
+/// 2. Convert matching flattened doc ids back to segment doc ids.
+/// 3. Apply any remaining row-level WHERE filter and materialize DISTINCT results, including missing-path handling.
public class JsonIndexDistinctOperator extends BaseOperator<DistinctResultsBlock> {
private static final String EXPLAIN_NAME = "DISTINCT_JSON_INDEX";
private static final String FUNCTION_NAME = "jsonExtractIndex";
+ /// Returns true if the expression is a `jsonExtractIndex` function call. All other validation (argument count/types,
+ /// column existence, JSON index presence, path support) happens inside the operator's constructor and matches what
+ /// the scan-based fallback (`JsonExtractIndexTransformFunction`) would surface during its own `init`.
+ public static boolean canUseJsonIndexDistinct(ExpressionContext expr) {
+ return expr.getType() == ExpressionContext.Type.FUNCTION && FUNCTION_NAME.equalsIgnoreCase(
+ expr.getFunction().getFunctionName());
+ }
+
private final IndexSegment _indexSegment;
- private final SegmentContext _segmentContext;
+ private final int _totalDocs;
private final QueryContext _queryContext;
private final BaseFilterOperator _filterOperator;
+ private final ExpressionContext _expression;
+ private final boolean _skipMissingPath;
+ private final JsonIndexReader _jsonIndexReader;
+ private final String _jsonPathString;
+ private final DataType _dataType;
+ @Nullable
+ private final String[] _defaultValueLiterals;
+ @Nullable
+ private final String _filterJsonExpression;
+ private final DataSchema _dataSchema;
+ @Nullable
+ private final OrderByExpressionContext _orderByExpression;
- private int _numEntriesExamined = 0;
+ private int _numDocsScanned = 0;
private long _numEntriesScannedInFilter = 0;
+ private int _numEntriesExaminedPostFilter = 0;
- public JsonIndexDistinctOperator(IndexSegment indexSegment, SegmentContext segmentContext,
- QueryContext queryContext, BaseFilterOperator filterOperator) {
+ public JsonIndexDistinctOperator(IndexSegment indexSegment, QueryContext queryContext,
+ BaseFilterOperator filterOperator) {
_indexSegment = indexSegment;
- _segmentContext = segmentContext;
+ _totalDocs = indexSegment.getSegmentMetadata().getTotalDocs();
_queryContext = queryContext;
_filterOperator = filterOperator;
- }
-
- @Override
- protected DistinctResultsBlock getNextBlock() {
- List<ExpressionContext> expressions = _queryContext.getSelectExpressions();
+ List<ExpressionContext> expressions = queryContext.getSelectExpressions();
if (expressions.size() != 1) {
throw new IllegalStateException("JsonIndexDistinctOperator supports single expression only");
}
+ _expression = expressions.get(0);
+ _skipMissingPath = QueryOptionsUtils.isJsonIndexDistinctSkipMissingPath(queryContext.getQueryOptions());
- ExpressionContext expr = expressions.get(0);
- ParsedJsonExtractIndex parsed = parseJsonExtractIndex(expr);
- if (parsed == null) {
- throw new IllegalStateException("Expected 3/4-arg scalar jsonExtractIndex expression");
- }
+ // Mirrors the arguments handling logic in `JsonExtractIndexTransformFunction`
- DataSource dataSource = _indexSegment.getDataSource(parsed._columnName, _queryContext.getSchema());
- JsonIndexReader jsonIndexReader = getJsonIndexReader(dataSource);
- if (jsonIndexReader == null) {
- throw new IllegalStateException("Column " + parsed._columnName + " has no JSON index");
+ List<ExpressionContext> arguments = _expression.getFunction().getArguments();
+ int numArguments = arguments.size();
+ // Check that there are exactly 3 or 4 or 5 arguments
+ if (numArguments < 3 || numArguments > 5) {
+ throw new IllegalArgumentException(
+ "Expected 3/4/5 arguments for jsonExtractIndex(jsonFieldName, 'jsonPath', 'resultsType',"
+ + " ['defaultValue'], ['jsonFilterExpression'])");
}
- String pushedDownFilterJson = extractSamePathJsonMatchFilter(parsed, _queryContext.getFilter());
- boolean filterFullyPushedDown = pushedDownFilterJson != null
- && isOnlySamePathJsonMatchFilter(parsed, _queryContext.getFilter())
- && !jsonMatchFilterCanMatchMissingPath(pushedDownFilterJson);
-
- // Fast path: when the filter is fully pushed down into the JSON index, we only need the distinct value strings.
- // This avoids reading posting lists, building per-value bitmaps, and converting flattened doc IDs.
- if (filterFullyPushedDown) {
- Set<String> distinctValues = jsonIndexReader.getMatchingDistinctValues(
- parsed._jsonPathString, pushedDownFilterJson);
- return buildDistinctResultsFromValues(expr, parsed, distinctValues);
+ ExpressionContext firstArgument = arguments.get(0);
+ if (firstArgument.getType() == ExpressionContext.Type.IDENTIFIER) {
+ DataSource dataSource = indexSegment.getDataSource(firstArgument.getIdentifier());
+ _jsonIndexReader = getJsonIndexReader(dataSource);
+ if (_jsonIndexReader == null) {
+ throw new IllegalStateException("jsonExtractIndex can only be applied on a column with JSON index");
+ }
+ } else {
+ throw new IllegalArgumentException("jsonExtractIndex can only be applied to a raw column");
}
- // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
- RoaringBitmap filteredDocIds = buildFilteredDocIds();
- if (filteredDocIds != null && filteredDocIds.isEmpty()) {
- ColumnDataType earlyColumnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema earlyDataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{earlyColumnDataType});
- OrderByExpressionContext earlyOrderBy = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- return new DistinctResultsBlock(
- createDistinctTable(earlyDataSchema, parsed._dataType, earlyOrderBy), _queryContext);
+ ExpressionContext secondArgument = arguments.get(1);
+ if (secondArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path argument must be a literal");
+ }
+ _jsonPathString = secondArgument.getLiteral().getStringValue();
+ try {
+ JsonPathCache.INSTANCE.getOrCompute(_jsonPathString);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("JSON path argument is not a valid JSON path");
}
- // All other WHERE filters remain row-level and are applied after converting flattened doc IDs to real doc IDs.
- Map<String, RoaringBitmap> valueToMatchingDocs =
- jsonIndexReader.getMatchingFlattenedDocsMap(parsed._jsonPathString, pushedDownFilterJson);
+ ExpressionContext thirdArgument = arguments.get(2);
+ if (thirdArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Result type argument must be a literal");
+ }
+ String resultsType = thirdArgument.getLiteral().getStringValue().toUpperCase();
+ boolean isSingleValue = !resultsType.endsWith("_ARRAY");
+ if (isSingleValue && _jsonPathString.contains("[*]")) {
+ throw new IllegalArgumentException(
+ "[*] syntax in json path is unsupported for singleValue field json_extract_index");
+ }
+ String dataTypeName = isSingleValue ? resultsType : resultsType.substring(0, resultsType.length() - 6);
+ try {
+ _dataType = DataType.valueOf(dataTypeName);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException("Unknown jsonExtractIndex result type: " + resultsType);
+ }
+ switch (_dataType) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case BIG_DECIMAL:
+ case STRING:
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported jsonExtractIndex result type for distinct: " + _dataType);
+ }
- // Always single-value (MV _ARRAY is rejected in parseJsonExtractIndex)
- jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
- return buildDistinctResultsBlock(expr, parsed, valueToMatchingDocs, filteredDocIds,
- filteredDocIds == null);
- }
+ // With _skipMissingPath, the 4-arg default is never used at runtime (handleMissingDocs is bypassed), so don't
+ // parse or validate it — accept any literal shape and ignore it.
+ if (numArguments >= 4 && !_skipMissingPath) {
+ ExpressionContext fourthArgument = arguments.get(3);
+ if (fourthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("Default value must be a literal");
+ }
+ String defaultLiteral = fourthArgument.getLiteral().getStringValue();
+ if (isSingleValue) {
+ try {
+ _dataType.convert(defaultLiteral);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + defaultLiteral + "' is not a valid " + _dataType);
+ }
+ _defaultValueLiterals = new String[]{defaultLiteral};
+ } else {
+ try {
+ JsonNode mvArray = JsonUtils.stringToJsonNode(defaultLiteral);
+ if (!mvArray.isArray()) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ String[] literals = new String[mvArray.size()];
+ for (int i = 0; i < mvArray.size(); i++) {
+ literals[i] = mvArray.get(i).asText();
+ try {
+ _dataType.convert(literals[i]);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Default value '" + literals[i] + "' is not a valid " + _dataType);
+ }
+ }
+ _defaultValueLiterals = literals;
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Default value must be a valid JSON array");
+ }
+ }
+ } else {
+ _defaultValueLiterals = null;
+ }
- private DistinctResultsBlock buildDistinctResultsFromValues(ExpressionContext expr, ParsedJsonExtractIndex parsed,
- Set<String> distinctValues) {
- ColumnDataType columnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema dataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{columnDataType});
- OrderByExpressionContext orderByExpression = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- DistinctTable distinctTable = createDistinctTable(dataSchema, parsed._dataType, orderByExpression);
- int limit = _queryContext.getLimit();
+ if (numArguments == 5) {
+ ExpressionContext fifthArgument = arguments.get(4);
+ if (fifthArgument.getType() != ExpressionContext.Type.LITERAL) {
+ throw new IllegalArgumentException("JSON path filter argument must be a literal");
+ }
+ _filterJsonExpression = fifthArgument.getLiteral().getStringValue();
+ } else {
+ _filterJsonExpression = null;
+ }
- for (String value : distinctValues) {
- _numEntriesExamined++;
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExamined, EXPLAIN_NAME);
+ _dataSchema = new DataSchema(new String[]{_expression.toString()},
+ new ColumnDataType[]{ColumnDataType.fromDataTypeSV(_dataType)});
+ List<OrderByExpressionContext> orderByExpressions = queryContext.getOrderByExpressions();
+ _orderByExpression = orderByExpressions != null ? orderByExpressions.get(0) : null;
+ }
- boolean done = addValueToDistinctTable(distinctTable, value, parsed._dataType, orderByExpression);
- if (done) {
- break;
- }
- if (orderByExpression == null && distinctTable.hasLimit() && distinctTable.size() >= limit) {
- break;
+ @Nullable
+ private static JsonIndexReader getJsonIndexReader(DataSource dataSource) {
+ JsonIndexReader reader = dataSource.getJsonIndex();
+ // TODO: rework
+ if (reader == null) {
+ Optional<IndexType<?, ?, ?>> compositeIndex = IndexService.getInstance().getOptional("composite_json_index");
+ if (compositeIndex.isPresent()) {
+ reader = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
}
}
+ return reader;
+ }
- return new DistinctResultsBlock(distinctTable, _queryContext);
+ @Override
+ protected DistinctResultsBlock getNextBlock() {
+ // Evaluate the filter first so we can skip the (potentially expensive) index map when no docs match.
+ BaseFilterOperator.FilteredDocIds filteredDocIds = _filterOperator.getFilteredDocIds();
+ ImmutableRoaringBitmap docIds = filteredDocIds.getDocIds();
+ _numDocsScanned = docIds != null ? docIds.getCardinality() : _totalDocs;
+ _numEntriesScannedInFilter = filteredDocIds.getNumEntriesScannedInFilter();
+ if (_numDocsScanned == 0) {
+ return new DistinctResultsBlock(createDistinctTable(), _queryContext);
+ }
+
+ // The 5-arg form's filter literal is pushed into the JSON index; WHERE-clause filters remain row-level and are
+ // applied after converting flattened doc IDs to real doc IDs.
+ Map<String, RoaringBitmap> valueToMatchingDocs =
+ _jsonIndexReader.getMatchingFlattenedDocsMap(_jsonPathString, _filterJsonExpression);
+ _jsonIndexReader.convertFlattenedDocIdsToDocIds(valueToMatchingDocs);
+ return buildDistinctResultsBlock(valueToMatchingDocs, docIds != null ? docIds.toRoaringBitmap() : null);
}
- private DistinctResultsBlock buildDistinctResultsBlock(ExpressionContext expr, ParsedJsonExtractIndex parsed,
- Map<String, RoaringBitmap> valueToMatchingDocs, @Nullable RoaringBitmap filteredDocIds,
- boolean allDocsSelected) {
- ColumnDataType columnDataType = ColumnDataType.fromDataTypeSV(parsed._dataType);
- DataSchema dataSchema = new DataSchema(
- new String[]{expr.toString()},
- new ColumnDataType[]{columnDataType});
- OrderByExpressionContext orderByExpression = _queryContext.getOrderByExpressions() != null
- ? _queryContext.getOrderByExpressions().get(0) : null;
- DistinctTable distinctTable = createDistinctTable(dataSchema, parsed._dataType, orderByExpression);
+ private DistinctResultsBlock buildDistinctResultsBlock(Map<String, RoaringBitmap> valueToMatchingDocs,
+ @Nullable RoaringBitmap filteredDocIds) {
+ DistinctTable distinctTable = createDistinctTable();
- int limit = _queryContext.getLimit();
- int totalDocs = _indexSegment.getSegmentMetadata().getTotalDocs();
- RoaringBitmap coveredDocs = allDocsSelected ? new RoaringBitmap() : null;
- RoaringBitmap remainingDocs = filteredDocIds != null ? filteredDocIds.clone() : null;
- boolean allDocsCovered = filteredDocIds == null ? !allDocsSelected || totalDocs == 0 : filteredDocIds.isEmpty();
+ // With _skipMissingPath, handleMissingDocs is bypassed — no need to track which docs are still uncovered, so
+ // skip the bitmap allocation and per-iteration `andNot` work entirely.
+ boolean allDocsCovered = _skipMissingPath;
+ RoaringBitmap remainingDocs = _skipMissingPath ? null
+ : (filteredDocIds != null ? filteredDocIds.clone() : RoaringBitmap.bitmapOfRange(0L, _totalDocs));
boolean earlyBreak = false;
for (Map.Entry<String, RoaringBitmap> entry : valueToMatchingDocs.entrySet()) {
- _numEntriesExamined++;
- QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExamined, EXPLAIN_NAME);
+ QueryThreadContext.checkTerminationAndSampleUsagePeriodically(_numEntriesExaminedPostFilter++, EXPLAIN_NAME);
String value = entry.getKey();
RoaringBitmap docIds = entry.getValue();
- boolean includeValue;
- if (filteredDocIds == null) {
- includeValue = true;
- if (!allDocsCovered && allDocsSelected) {
- coveredDocs.or(docIds);
- if (coveredDocs.getLongCardinality() >= totalDocs) {
- allDocsCovered = true;
- }
- }
- } else {
- includeValue = RoaringBitmap.intersects(docIds, filteredDocIds);
- // Remove matched docs from remaining set in-place (no allocation per value).
- if (!allDocsCovered && includeValue) {
- remainingDocs.andNot(docIds);
- if (remainingDocs.isEmpty()) {
- allDocsCovered = true;
- }
- }
- }
+ // Unfiltered always includes; filtered must intersect the original filter set (not the shrinking
+ // `remainingDocs`, since a value can still belong to the result after all filtered docs are covered).
+ boolean includeValue = filteredDocIds == null || RoaringBitmap.intersects(docIds, filteredDocIds);
- if (includeValue) {
- boolean done = addValueToDistinctTable(distinctTable, value, parsed._dataType, orderByExpression);
- if (done) {
- earlyBreak = true;
- break;
+ if (!allDocsCovered && includeValue) {
+ remainingDocs.andNot(docIds);
+ if (remainingDocs.isEmpty()) {
+ allDocsCovered = true;
}
}
- if (orderByExpression == null && distinctTable.hasLimit() && distinctTable.size() >= limit) {
+ // addValueToDistinctTable returns true exactly when the table has reached its LIMIT (no-ORDER-BY case);
+ // for ORDER-BY or unbounded LIMIT it always returns false. So no separate size check is needed.
+ if (includeValue && addValueToDistinctTable(distinctTable, value)) {
earlyBreak = true;
break;
}
}
if (!earlyBreak && !allDocsCovered) {
- handleMissingDocs(distinctTable, parsed, orderByExpression);
+ handleMissingDocs(distinctTable);
}
return new DistinctResultsBlock(distinctTable, _queryContext);
}
- private void handleMissingDocs(DistinctTable distinctTable, ParsedJsonExtractIndex parsed,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (parsed._defaultValueLiteral != null) {
- addValueToDistinctTable(distinctTable, parsed._defaultValueLiteral, parsed._dataType, orderByExpression);
- } else if (_queryContext.isNullHandlingEnabled()) {
- distinctTable.addNull();
- } else {
- throw new RuntimeException(
- String.format("Illegal Json Path: [%s], for some docIds in segment [%s]",
- parsed._jsonPathString, _indexSegment.getSegmentName()));
- }
- }
-
- @Nullable
- private static String extractSamePathJsonMatchFilter(ParsedJsonExtractIndex parsed, @Nullable FilterContext filter) {
- if (filter == null) {
- return null;
- }
- switch (filter.getType()) {
- case PREDICATE:
- return extractSamePathJsonMatchFilter(parsed, filter.getPredicate());
- case AND:
- String matchingFilter = null;
- for (FilterContext child : filter.getChildren()) {
- String childFilter = extractSamePathJsonMatchFilter(parsed, child);
- if (childFilter == null) {
- continue;
- }
- if (matchingFilter != null) {
- return null;
- }
- matchingFilter = childFilter;
- }
- return matchingFilter;
- default:
- return null;
- }
- }
-
- private static boolean isOnlySamePathJsonMatchFilter(ParsedJsonExtractIndex parsed, @Nullable FilterContext filter) {
- if (filter == null || filter.getType() != FilterContext.Type.PREDICATE) {
- return false;
- }
- return extractSamePathJsonMatchFilter(parsed, filter.getPredicate()) != null;
- }
-
- private static boolean jsonMatchFilterCanMatchMissingPath(String filterJsonString) {
- try {
- FilterContext filter = RequestContextUtils.getFilter(CalciteSqlParser.compileToExpression(filterJsonString));
- return filter.getType() == FilterContext.Type.PREDICATE
- && filter.getPredicate().getType() == Predicate.Type.IS_NULL;
- } catch (Exception e) {
- return false;
- }
- }
-
- @Nullable
- private static String extractSamePathJsonMatchFilter(ParsedJsonExtractIndex parsed, Predicate predicate) {
- if (!(predicate instanceof JsonMatchPredicate)) {
- return null;
- }
- ExpressionContext lhs = predicate.getLhs();
- if (lhs.getType() != ExpressionContext.Type.IDENTIFIER
- || !parsed._columnName.equals(lhs.getIdentifier())) {
- return null;
- }
- String filterJsonString = ((JsonMatchPredicate) predicate).getValue();
- int start = filterJsonString.indexOf('"');
- if (start < 0) {
- return null;
- }
- int end = filterJsonString.indexOf('"', start + 1);
- if (end < 0) {
- return null;
- }
- String filterPath = filterJsonString.substring(start + 1, end);
- return parsed._jsonPathString.equals(filterPath) ? filterJsonString : null;
- }
-
- private DistinctTable createDistinctTable(DataSchema dataSchema, FieldSpec.DataType dataType,
- @Nullable OrderByExpressionContext orderByExpression) {
+ private DistinctTable createDistinctTable() {
int limit = _queryContext.getLimit();
boolean nullHandlingEnabled = _queryContext.isNullHandlingEnabled();
- switch (dataType) {
+ switch (_dataType) {
case INT:
- return new IntDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new IntDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case LONG:
- return new LongDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new LongDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case FLOAT:
- return new FloatDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new FloatDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case DOUBLE:
- return new DoubleDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new DoubleDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case BIG_DECIMAL:
- return new BigDecimalDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new BigDecimalDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
case STRING:
- return new StringDistinctTable(dataSchema, limit, nullHandlingEnabled, orderByExpression);
+ return new StringDistinctTable(_dataSchema, limit, nullHandlingEnabled, _orderByExpression);
default:
- throw new IllegalStateException("Unsupported data type for JSON index distinct: " + dataType);
+ throw new IllegalStateException("Unsupported data type for JSON index distinct: " + _dataType);
}
}
- private static boolean addValueToDistinctTable(DistinctTable distinctTable, String stringValue,
- FieldSpec.DataType dataType, @Nullable OrderByExpressionContext orderByExpression) {
- switch (dataType) {
+ private boolean addValueToDistinctTable(DistinctTable distinctTable, String stringValue) {
+ switch (_dataType) {
case INT:
- return addToTable((IntDistinctTable) distinctTable, Integer.parseInt(stringValue), orderByExpression);
+ return addToTable((IntDistinctTable) distinctTable, Integer.parseInt(stringValue));
case LONG:
- return addToTable((LongDistinctTable) distinctTable, Long.parseLong(stringValue), orderByExpression);
+ return addToTable((LongDistinctTable) distinctTable, Long.parseLong(stringValue));
case FLOAT:
- return addToTable((FloatDistinctTable) distinctTable, Float.parseFloat(stringValue), orderByExpression);
+ return addToTable((FloatDistinctTable) distinctTable, Float.parseFloat(stringValue));
case DOUBLE:
- return addToTable((DoubleDistinctTable) distinctTable, Double.parseDouble(stringValue), orderByExpression);
+ return addToTable((DoubleDistinctTable) distinctTable, Double.parseDouble(stringValue));
case BIG_DECIMAL:
- return addToTable((BigDecimalDistinctTable) distinctTable, new BigDecimal(stringValue), orderByExpression);
+ return addToTable((BigDecimalDistinctTable) distinctTable, new BigDecimal(stringValue));
case STRING:
- return addToTable((StringDistinctTable) distinctTable, stringValue, orderByExpression);
+ return addToTable((StringDistinctTable) distinctTable, stringValue);
default:
- throw new IllegalStateException("Unsupported data type for JSON index distinct: " + dataType);
+ throw new IllegalStateException("Unsupported data type for JSON index distinct: " + _dataType);
}
}
- private static boolean addToTable(IntDistinctTable table, int value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(IntDistinctTable table, int value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
- }
-
- private static boolean addToTable(LongDistinctTable table, long value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
- table.addUnbounded(value);
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
return false;
}
+ return table.addWithoutOrderBy(value);
}
- private static boolean addToTable(FloatDistinctTable table, float value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(LongDistinctTable table, long value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
- }
-
- private static boolean addToTable(DoubleDistinctTable table, double value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
- table.addUnbounded(value);
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
return false;
}
+ return table.addWithoutOrderBy(value);
}
- private static boolean addToTable(BigDecimalDistinctTable table, BigDecimal value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(FloatDistinctTable table, float value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
+ }
+ return table.addWithoutOrderBy(value);
}
- private static boolean addToTable(StringDistinctTable table, String value,
- @Nullable OrderByExpressionContext orderByExpression) {
- if (table.hasLimit()) {
- if (orderByExpression != null) {
- table.addWithOrderBy(value);
- return false;
- } else {
- return table.addWithoutOrderBy(value);
- }
- } else {
+ private boolean addToTable(DoubleDistinctTable table, double value) {
+ if (!table.hasLimit()) {
table.addUnbounded(value);
return false;
}
- }
-
- @Nullable
- private static JsonIndexReader getJsonIndexReader(DataSource dataSource) {
- JsonIndexReader reader = dataSource.getJsonIndex();
- if (reader == null) {
- Optional<IndexType<?, ?, ?>> compositeIndex =
- IndexService.getInstance().getOptional("composite_json_index");
- if (compositeIndex.isPresent()) {
- reader = (JsonIndexReader) dataSource.getIndex(compositeIndex.get());
- }
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
}
- return reader;
- }
-
- @Nullable
- private RoaringBitmap buildFilteredDocIds() {
- BaseFilterOperator.FilteredDocIds filteredDocIds = _filterOperator.getFilteredDocIds();
- _numEntriesScannedInFilter = filteredDocIds.getNumEntriesScannedInFilter();
- ImmutableRoaringBitmap docIds = filteredDocIds.getDocIds();
- return docIds != null ? docIds.toRoaringBitmap() : null;
+ return table.addWithoutOrderBy(value);
}
- @Nullable
- private static ParsedJsonExtractIndex parseJsonExtractIndex(ExpressionContext expr) {
- if (expr.getType() != ExpressionContext.Type.FUNCTION) {
- return null;
- }
- if (!FUNCTION_NAME.equalsIgnoreCase(expr.getFunction().getFunctionName())) {
- return null;
- }
- List<ExpressionContext> args = expr.getFunction().getArguments();
- if (args.size() != 3 && args.size() != 4) {
- return null;
- }
- if (args.get(0).getType() != ExpressionContext.Type.IDENTIFIER) {
- return null;
- }
- if (args.get(1).getType() != ExpressionContext.Type.LITERAL
- || args.get(2).getType() != ExpressionContext.Type.LITERAL
- || (args.size() == 4 && args.get(3).getType() != ExpressionContext.Type.LITERAL)) {
- return null;
- }
-
- String columnName = args.get(0).getIdentifier();
- String jsonPathString = args.get(1).getLiteral().getStringValue();
- String resultsType = args.get(2).getLiteral().getStringValue().toUpperCase();
- // Only single-value types are supported; MV (_ARRAY) would have incorrect flattened-to-real
- // docId intersection since convertFlattenedDocIdsToDocIds is skipped for MV.
- if (resultsType.endsWith("_ARRAY")) {
- return null;
- }
- if (jsonPathString.contains("[*]")) {
- return null;
- }
-
- FieldSpec.DataType dataType;
- try {
- dataType = FieldSpec.DataType.valueOf(resultsType);
- } catch (IllegalArgumentException e) {
- return null;
+ private boolean addToTable(BigDecimalDistinctTable table, BigDecimal value) {
+ if (!table.hasLimit()) {
+ table.addUnbounded(value);
+ return false;
}
- // Only types with a corresponding DistinctTable implementation are supported
- switch (dataType) {
- case INT:
- case LONG:
- case FLOAT:
- case DOUBLE:
- case BIG_DECIMAL:
- case STRING:
- break;
- default:
- return null;
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
}
+ return table.addWithoutOrderBy(value);
+ }
- try {
- JsonPathCache.INSTANCE.getOrCompute(jsonPathString);
- } catch (Exception e) {
- return null;
+ private boolean addToTable(StringDistinctTable table, String value) {
+ if (!table.hasLimit()) {
+ table.addUnbounded(value);
+ return false;
}
-
- String defaultValueLiteral = null;
- if (args.size() == 4) {
- defaultValueLiteral = args.get(3).getLiteral().getStringValue();
- try {
- dataType.convert(defaultValueLiteral);
- } catch (Exception e) {
- return null;
- }
+ if (_orderByExpression != null) {
+ table.addWithOrderBy(value);
+ return false;
}
-
- return new ParsedJsonExtractIndex(columnName, jsonPathString, dataType, defaultValueLiteral);
+ return table.addWithoutOrderBy(value);
}
- private static final class ParsedJsonExtractIndex {
- final String _columnName;
- final String _jsonPathString;
- final FieldSpec.DataType _dataType;
- @Nullable
- final String _defaultValueLiteral;
-
- ParsedJsonExtractIndex(String columnName, String jsonPathString, FieldSpec.DataType dataType,
- @Nullable String defaultValueLiteral) {
- _columnName = columnName;
- _jsonPathString = jsonPathString;
- _dataType = dataType;
- _defaultValueLiteral = defaultValueLiteral;
+ private void handleMissingDocs(DistinctTable distinctTable) {
+ if (_defaultValueLiterals != null) {
+ for (String literal : _defaultValueLiterals) {
+ if (addValueToDistinctTable(distinctTable, literal)) {
+ return;
+ }
+ }
+ } else if (_queryContext.isNullHandlingEnabled()) {
+ distinctTable.addNull();
+ } else {
+ throw new RuntimeException(
+ String.format("Illegal Json Path: [%s], for some docIds in segment [%s]", _jsonPathString,
Review Comment:
This aligns with the behavior in `JsonExtractIndexTransformFunction`. The behavior should align when using index based distinct vs scan based fallback
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Refine index-based DISTINCT operators (JSON / inverted) [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #18588:
URL: https://github.com/apache/pinot/pull/18588
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org