[multistage] add maxRowsInJoin, maxRowsInWindow, numGroups to query response

[dang-stripe]

summary

this addresses #17565

• this adds maxRowsInJoin, maxRowsInWindow, numGroups to the query response so we can monitor how close queries are to their respective limits before queries start failing (throw overflow mode) or returning partial results (break overflow mode)
• these fields are only updated for multistage. it does not update numGroups for single stage.

cc @Jackie-Jiang @gortiz @suvodeep-pyne

testing

• added unit tests to cover edge cases
• ran queries on a test cluster + table to verify fields return intended values. note that exchange_rates is a dim table.

#	Query Type	Mode	Query	maxRowsInJoin	numGroups	maxRowsInWindow	Error?	Result
Baseline
1	Simple SELECT	normal	SELECT order_id, amount, currency FROM orders ORDER BY order_id LIMIT 3	0	0	0	—	PASS
JOIN
2	Hash JOIN	normal	SELECT o.order_id, o.amount, r.rate FROM orders AS o JOIN exchange_rates AS r ON o.currency = r.currency LIMIT 5	4	0	0	—	PASS
3	Lookup JOIN	normal	SELECT /*+ joinOptions(join_strategy='lookup') */ o.order_id, o.amount, r.rate FROM orders AS o JOIN exchange_rates AS r ON o.currency = r.currency LIMIT 5	0 (lookup bypasses HashJoinOperator — no hash table built)	0	0	—	PASS
4	Hash JOIN	BREAK (limit=10)	SELECT /*+ joinOptions( max_rows_in_join='10', join_overflow_mode='BREAK') */ o.order_id, o.amount, r.rate FROM orders AS o JOIN exchange_rates AS r ON o.currency = r.currency LIMIT 5	10 (limit reached)	0	0	—	PASS
5	Hash JOIN (self-join)	THROW (limit=10)	SELECT /*+ joinOptions( max_rows_in_join='10', join_overflow_mode='THROW') */ a.order_id, b.amount FROM orders AS a JOIN orders AS b ON a.customer_id = b.customer_id LIMIT 5	2525	0	0	245 — resource limit exceeded	PASS
GROUP BY
6	GROUP BY (high card.)	normal	SELECT order_id, COUNT(*) as cnt FROM orders GROUP BY order_id ORDER BY cnt DESC LIMIT 5	0	2505	0	—	PASS
7	GROUP BY (low card.)	normal	SELECT currency, COUNT(*) as cnt, SUM(amount) as total_amount FROM orders GROUP BY currency	0	1	0	—	PASS
8	Global aggregate	normal	SELECT SUM(amount) as total, AVG(amount) as avg_amount FROM orders	0	0	0	—	PASS
9	GROUP BY	BREAK (limit=100)	SELECT /*+ aggOptions( num_groups_limit='100') */ order_id, COUNT(*) as cnt FROM orders GROUP BY order_id ORDER BY cnt DESC LIMIT 5	0	100 (limit reached)	0	—	PASS
10	GROUP BY	THROW (limit=10)	SELECT /*+ aggOptions( num_groups_limit='10', error_on_num_groups_limit='true') */ order_id, COUNT(*) as cnt FROM orders GROUP BY order_id ORDER BY cnt DESC LIMIT 5	0	10	0	1000 — query execution error	PASS
Window
11	Window ROW_NUMBER()	normal	SELECT order_id, amount, ROW_NUMBER() OVER (ORDER BY amount DESC) as rn FROM orders LIMIT 5	0	0	5000	—	PASS
12	Window	BREAK (limit=100)	SET maxRowsInWindow = 100; SET windowOverflowMode = 'BREAK'; SELECT order_id, amount, ROW_NUMBER() OVER (ORDER BY amount DESC) as rn FROM orders LIMIT 5	0	0	100 (limit reached)	—	PASS
13	Window	THROW (limit=10)	SET maxRowsInWindow = 10; SELECT order_id, amount, ROW_NUMBER() OVER (ORDER BY amount DESC) as rn FROM orders LIMIT 5	0	0	5000	245 — resource limit exceeded	PASS
Combined operators
14	JOIN + GROUP BY	normal	SELECT r.currency, COUNT(*) as cnt, SUM(o.amount) as total FROM orders AS o JOIN exchange_rates AS r ON o.currency = r.currency GROUP BY r.currency	3	1	0	—	PASS
15	JOIN + GROUP BY + Window	normal	WITH grouped AS ( SELECT o.currency, r.rate, COUNT(*) as cnt, SUM(o.amount) as total FROM orders AS o JOIN exchange_rates AS r ON o.currency = r.currency GROUP BY o.currency, r.rate ) SELECT currency, rate, cnt, total, ROW_NUMBER() OVER (ORDER BY total DESC) as rn FROM grouped	3	1	1	—	PASS

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.31034% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 63.27%. Comparing base (325ac2a) to head (49cc67f).
⚠️ Report is 80 commits behind head on master.

Files with missing lines	Patch %	Lines
...common/response/broker/BrokerResponseNativeV2.java	58.33%	5 Missing ⚠️
...not/query/runtime/operator/MultiStageOperator.java	66.66%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #17784      +/-   ##
============================================
+ Coverage     63.17%   63.27%   +0.09%     
- Complexity     1454     1460       +6     
============================================
  Files          3176     3190      +14     
  Lines        191025   192067    +1042     
  Branches      29206    29421     +215     
============================================
+ Hits         120688   121530     +842     
- Misses        60920    61024     +104     
- Partials       9417     9513      +96     

Flag	Coverage Δ
custom-integration1	?
integration	0.00% <ø> (-100.00%)	⬇️
integration1	?
integration2	0.00% <ø> (ø)
java-11	63.23% <79.31%> (+0.07%)	⬆️
java-21	63.24% <79.31%> (+0.10%)	⬆️
temurin	63.27% <79.31%> (+0.09%)	⬆️
unittests	63.27% <79.31%> (+0.09%)	⬆️
unittests1	55.60% <79.31%> (+0.06%)	⬆️
unittests2	34.25% <0.00%> (+0.19%)	⬆️

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

☔ View full report in Codecov by Sentry.
📢 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.

[gortiz]

I would like to look for other approaches instead of this one. I have the feeling we are bloating up the response with too many attributes that are either circumstancial or derivable from the query stats.

I understand this means to move the processing to the client side, which may not be the best, but we can have other alternatives like some optional response decorator that takes tle stageStats and create new attributes based on that. With a system like this you can create your own decorators without increasing the already too large metadata response

[dang-stripe]

@gortiz replicating the logic for how these limits are enforced on the client seems like a significant burden. i believe stage stats is not a stable API yet either meaning this can break across releases. if that is the expectation, ideally there should be a canonical library for processing stage stats.

i feel that the current approach supplements the existing fields in the query response (maxRowsInJoinReached, maxRowsInWindowReached) well without having to understand the complexity of stage stats.

[gortiz]

I believe stage stats is not a stable API yet either meaning this can break across releases.

I wouldn't say so. That API is stable and publicly documented (see each operator in https://docs.pinot.apache.org/users/user-guide-query/multi-stage-query/operator-types). There may be cases where the broker cannot reconstruct stageStats, and that is something we want to improve, but I don't recall any case where stats were renamed or removed. In fact, removing entries from the Java enum is explicitly forbidden.

i feel that the current approach supplements the existing fields in the query response (maxRowsInJoinReached, maxRowsInWindowReached) well without having to understand the complexity of stage stats.

Historically, we added a new entry in the response for whatever requirement someone has. As a result, today we have dozens of entries that are not very well-defined and verbose in code (requiring getters and setters for each new property).

[gortiz]

Approving it as it has been proven useful for Stripe. Before merging it, @dang-stripe can you clean up the code so we don't include the getters to the interface and BrokerResponseNative (V1)?

[gortiz]

Do we need to add these methods to the interface? I don't see any call to this, getMaxRowsInWindow or getNumGroups and it is not actually necessary for jackson.

For example, getMaxRowsInOperator() is defined only for BrokerNativeV2.

[dang-stripe]

Thanks @gortiz! I've updated the PR to remove the fields from the interface and V1 response. Re-tested it and confirmed no change.