Skip to content

Support IN_SUBQUERY in the multi-stage engine and run IdSet subqueries first - #19804

Merged
yashmayya merged 3 commits into
apache:masterfrom
yashmayya:mse-idset-phase3
Oct 15, 2026
Merged

yashmayya merged 3 commits into
apache:masterfrom
yashmayya:mse-idset-phase3

Conversation

@yashmayya

@yashmayya yashmayya commented Oct 8, 2026 •

Copy link
Copy Markdown
Contributor

PR flow

The broker rewrites queries to run IdSet subqueries first, replacing them with IdSet literals before compilation.

flowchart TD
  N0["Receive query #40;F2#41;"]:::stUnchanged
  N1["Rewrite IdSet subqueries #40;F1#44; F2#41;"]:::stAdded
  N2["Compile rewritten query #40;F2#41;"]:::stUnchanged
  N3["Explain query and add extra column for IdSet subqueries not run #40;F2#41;"]:::stModified
  N4["Execute query #40;F2#41;"]:::stUnchanged
  N0 --> N1
  N1 --> N2
  N2 -->|"if EXPLAIN"| N3
  N2 -->|"if not EXPLAIN"| N4
  classDef stAdded fill:#dafbe1,stroke:#1a7f37,color:#1f2328,stroke-width:2px
  classDef stModified fill:#fff8c5,stroke:#9a6700,color:#1f2328,stroke-width:2px
  classDef stRemoved fill:#ffebe9,stroke:#cf222e,color:#1f2328,stroke-width:2px
  classDef stUnchanged fill:#f6f8fa,stroke:#656d76,color:#1f2328,stroke-width:1px
Loading

AI-generated · Green: added · Yellow: modified · Red: removed · Gray: existing

Partial evidence: 0 file patches omitted; 1 truncated.

Diff evidence
  • F1: pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/IdSetSubqueryRewriter.java — after
  • F2: pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java — before · after
  • Regenerate PR flow

Phase 3 of IdSet support in the multi-stage engine (MSE). Phases 1 (#19757) and 2 (#19759) are merged.

Problem

  • MSE rejects IN_SUBQUERY(col, 'subquery') with "No match found for function signature IN_SUBQUERY".
  • IN_ID_SET(col, (SELECT IDSET(...) ...)) has worked since Make IDSET and IN_ID_SET work in the multi-stage engine #19757, but slowly. The planner joins the subquery result to the rows, so every row leaves the leaf stage before the IdSet test runs.
  • To filter by a large set, a client therefore runs two queries: one gets the IdSet, and the other sends it back as a literal. Over HTTP, a literal can have at most 20M chars, which is about 12M ids at 1% false positives.

Changes

  • Before it compiles a query, the broker runs each IdSet subquery as a query of its own and puts the IdSet in its place. The planner and the servers then see IN_ID_SET(col, '<IdSet>'), so the test runs in the leaf stage. Servers and the wire format do not change. Two forms run first:
    • IN_SUBQUERY(expr, 'subquery'), anywhere in the query: SELECT, WHERE, HAVING, JOIN ON, nested queries, CTEs and set operations.
    • IN_ID_SET(expr, (subquery)), when the subquery validates on its own (i.e. is not correlated) and names no WITH item of the query. Other subqueries run as part of the query, as before.
  • Each subquery:
    • runs on MSE with a new request id, once even if it appears more than once.
    • gets the time left until the query's deadline.
    • inherits the query's options. SET statements in an IN_SUBQUERY string override them, and sqlOptionsMode applies to them. dropResults is not inherited.
    • runs like any other query: it is checked for table access, quota and throttling, it is listed as running, and the query event listener sees it.
    • can be cancelled with the query's client query id. The query then fails.
    • must return one STRING column and at most one row. No row, or NULL, gives the empty IdSet: = 1 is then false and = 0 true for every row.
    • fails the query if its result is partial, e.g. numGroupsLimitReached, because the IdSet would miss ids.
  • As in the single-stage engine, subqueries run before the query is checked for table access and quota.
  • EXPLAIN runs no subqueries. The plan uses the empty IdSet instead, and a new ID_SET_SUBQUERIES_NOT_RUN column lists the subqueries as a JSON array.
  • IN_SUBQUERY is registered in the MSE operator table. This lets paths that only compile queries accept it, such as the controller routing a query to a broker. As in the single-stage engine, those paths do not see the tables named in the subquery string.
  • New QueryEnvironment.validate(SqlNodeAndOptions): it validates without converting or optimizing, and reports every failure as a QueryException, including an invalid query option. The broker uses it to check whether a scalar subquery stands on its own. If that check fails for any reason, the subquery stays in the query, and the query's compilation reports the problem.
  • New RequestUtils.setOptions(SqlNodeAndOptions, Map) overload: it takes the request options as a map.

Behavior changes

  • The singleStageQueriesInvalidMultiStage meter no longer counts single-stage IN_SUBQUERY queries.
  • EXPLAIN of an uncorrelated IN_ID_SET(col, (subquery)) now shows the IdSet test in the leaf stage instead of a join, plus the new column. Make IDSET and IN_ID_SET work in the multi-stage engine #19757 is in no release yet.
  • partialResult is now true when a window stops at its row limit (overflow mode BREAK), as it already is for joins. The window flag was missing from isPartialResult(). The query logger now logs these queries like other partial results. The default mode, THROW, does not change.

Scale

In the integration test, a STRING IdSet sized for 20M ids (a 32M-char literal) filters the table in 0.5 s through IN_SUBQUERY. The set never reaches the client or the parser, so the 20M-char HTTP limit does not apply.

The limit is now the size of what the broker sends to each server:

  • Each place in the query that uses an IdSet carries its own copy of the literal, even when the subquery ran once.
  • The broker sends the plans of all the stages that run on a server in one gRPC message.
  • The MSE server accepts messages of up to 64 MB by default (pinot.query.multistage.query.server.max.inbound.message.size.bytes).

So one use of a 32M-char set fits, but two uses on the same server need a higher limit. Otherwise, dispatch fails with RESOURCE_EXHAUSTED.

Tests

  • IdSetSubqueryRewriterTest: every SQL position, nested queries, name variants, the text of scalar subqueries, correlated subqueries, WITH names, repeated subqueries, EXPLAIN and invalid calls.
  • MultiStageIdSetSubqueryTest: inherited options, identity and headers; time left; result checks, including partial results; errors; invalid inherited options; access checks on the subquery's tables; query events; and EXPLAIN.
  • QueryCompilationTest: queries with IN_SUBQUERY compile, and validate reports an invalid option as a QueryException.
  • IdSetIntegrationTest: IN_SUBQUERY in both engines, outside filters, and nested; scalar subqueries, both uncorrelated and correlated; a partial subquery result; and a 20M-id set.
  • BaseClusterIntegrationTestSet: the IN_ID_SET and IN_SUBQUERY hard-coded queries now run on both engines. The hybrid, realtime, logical table, cursor and MSE suites pass locally.
  • JmxMetricsIntegrationTest: now expects 2 invalid queries instead of 3.

Next

  • Optional phase 4, measure first: one shared Bloom filter per query and server while a set is built.
  • A clear error before dispatch when the plans for one server exceed its message limit, instead of RESOURCE_EXHAUSTED.
  • Single-stage IN_SUBQUERY throws an NPE when its subquery runs with null handling and matches no rows. A separate fix can also match the result checks above.

@yashmayya yashmayya added multi-stage Related to the multi-stage query engine feature New functionality labels Oct 8, 2026
@yashmayya
yashmayya requested a review from gortiz October 8, 2026 22:14
@codecov-commenter

codecov-commenter commented Oct 8, 2026 •

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.61%. Comparing base (f818bae) to head (947c303).
⚠️ Report is 51 commits behind head on master.

❗ There is a different number of reports uploaded between BASE (f818bae) and HEAD (947c303). Click for more details.

HEAD has 4 uploads less than BASE
Flag BASE (f818bae) HEAD (947c303)
java-25 6 5
temurin 6 5
unittests2 1 0
unittests 2 1
Additional details and impacted files
@@             Coverage Diff              @@
##             master   #19804      +/-   ##
============================================
- Coverage     68.55%   58.61%   -9.94%     
+ Complexity     1486        7    -1479     
============================================
  Files          3525     2722     -803     
  Lines        229852   169972   -59880     
  Branches      36507    27944    -8563     
============================================
- Hits         157568    99629   -57939     
- Misses        59965    61916    +1951     
+ Partials      12319     8427    -3892     
Flag Coverage Δ
integration 100.00% <ø> (ø)
integration1 100.00% <ø> (ø)
integration2 0.00% <ø> (ø)
java-25 58.61% <100.00%> (-9.94%) ⬇️
lane-a 100.00% <ø> (ø)
lane-b 0.00% <ø> (ø)
temurin 58.61% <100.00%> (-9.94%) ⬇️
unittests 58.61% <100.00%> (-9.94%) ⬇️
unittests1 58.61% <100.00%> (+0.30%) ⬆️
unittests2 ?

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

QueryEnvironment queryEnv = new QueryEnvironment(
getQueryEnvConf(httpHeaders, sqlNodeAndOptions.getOptions(), requestId), _multiClusterRoutingContext);
try {
queryEnv.validate(sqlNodeAndOptions);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

QueryEnvironment.validate(SqlNodeAndOptions) only wraps failures that happen inside validate(SqlNode, PlannerContext). The QueryEnvironment constructor and getPlannerContext() run before that try, so anything they throw because of an inherited option escapes here as a plain RuntimeException. The probe then fails the whole request as an UNKNOWN error, instead of returning false and keeping the join plan. Could we catch RuntimeException here and return false? The main compile still reports the real problem afterwards. The Javadoc of the new QueryEnvironment.validate also says it throws QueryException, which is not true for those paths, so either widen its try or adjust the doc.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch, fixed in 4db75a8. The check now returns false on any failure, so the subquery stays in the query and the main compile reports the real error. QueryEnvironment.validate also wraps those failures in a QueryException now, so its Javadoc holds. The time-left check stays outside the try, so a timeout still fails fast.

}
}

private String getIdSet(String subquery) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The cache runs a repeated subquery once, but each call site still gets its own copy of the literal in the plan. If the same 32 MB set appears twice in a stage that goes to one server, that StagePlan is about 64 MB, which is the default max inbound message size of the MSE QueryServer. So the practical ceiling is now the gRPC dispatch limit per server message, not the 20M-char HTTP limit. Could you mention that in the PR description and in the docs? A clear error before dispatch would be nicer than a gRPC RESOURCE_EXHAUSTED, but that can be a follow-up.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed. It is a bit wider than one stage: the broker sends the plans of all the stages a server runs in one message, so copies in different stages add up too. I added this to the PR description (Scale section), with the server key to raise. I will add it to the docs PR for the IdSet changes, and I listed the pre-dispatch error as a follow-up under Next.

@xiangfu0 xiangfu0 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reviewed the full diff with a focus on subquery semantics, planning, backward compatibility, and test coverage. This is a well-structured change: servers and the wire format never change (they only ever see IN_ID_SET(expr, '<literal>')), the EXPLAIN column is broker-local, and the test coverage is strong across the rewriter, the handler, compilation, and end-to-end.

I found no CRITICAL or MAJOR issues, so approving. A few minor observations are left inline for your confirmation — none are blocking.

Things I specifically verified and found sound:

  • Subquery text extraction re-parses and equalsDeep-compares, so any position/sanitization mismatch safely leaves the subquery in place rather than mis-rewriting.
  • WITH-item scoping, correlated detection (via validate), run-once dedup, and the empty-IdSet null/no-row semantics all behave as documented and are tested.
  • Per-table access control and table QPS quota still apply to each subquery; deadline propagation caps the subquery timeout to the time left.
  • The isPartialResult window-flag change only affects BREAK overflow mode; the default THROW mode is unchanged.

// The rows of the subquery are its result
options.remove(QueryOptionKey.DROP_RESULTS);

// Like BaseBrokerRequestHandler.handleRequest, except for the first access check and the application quota, which

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice that this is called out explicitly. Just to confirm the intent: because subqueries go through the 8-arg handleRequest rather than the public wrapper, the application-level QPS quota (acquireApplication) is consumed only once for the whole query, so a single accepted request with N IN_SUBQUERY calls fans out to N+1 executions on one app-quota token. That matches the single-stage engine, so this is probably intended — flagging only so it's a conscious decision.

/// cancelled. As it inherits the client query id, cancelling the query by that id cancels the subquery, which fails
/// the query.
@Nullable
private String runIdSetSubquery(long requestId, String subquery, Map<String, String> queryOptions,

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A subquery that itself contains IN_SUBQUERY recurses through handleRequest → handleRequestThrowing → rewriteIdSetSubqueries → here again. Depth is effectively bounded by the shared deadline and the SQL text, so it's not a real risk, but a small explicit max-depth guard would let a pathological nested query fail fast with a clear message instead of leaning on the timeout. Optional.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I'll defer this one

public boolean isPartialResult() {
return getExceptionsSize() > 0 || isNumGroupsLimitReached() || !getEarlyTerminationReasons().isEmpty()
|| isMaxRowsInJoinReached() || isMseLiteLeafStageLimitReached();
|| isMaxRowsInJoinReached() || isMaxRowsInWindowReached() || isMseLiteLeafStageLimitReached();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This now marks a response partial whenever the window row limit was hit. Worth confirming it only triggers in overflow mode BREAK (the default THROW throws instead and never sets the flag), so existing default-mode queries are unaffected. It mirrors the existing maxRowsInJoinReached handling and is required for the subquery partial-result guard, so this looks correct — just noting the client-visible partialResult behavior change for the record.

@yashmayya
yashmayya merged commit 49ce7be into apache:master Oct 15, 2026
14 checks passed
xiangfu0 added a commit to pinot-contrib/pinot-docs that referenced this pull request Oct 15, 2026
Follow-up to apache/pinot#19804. Updates engine compatibility and
explains eligible scalar subqueries, independent
checks/options/deadline, result shape, empty and partial results,
cancellation, and EXPLAIN placeholders.

---------

Co-authored-by: Xiang Fu <xiangfu@Xiang-mac-mtv-2.local>
@xiangfu0

Copy link
Copy Markdown
Contributor

Documentation follow-up merged: pinot-contrib/pinot-docs#1111 — MSE IN_SUBQUERY support and eligible IdSet pre-execution, options/deadline/checks, result/partial semantics, EXPLAIN, and plan-size caveats.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

feature New functionality multi-stage Related to the multi-stage query engine

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants