[multistage] Add Support for Values in Physical Optimizer

[ankitsultana]

Summary

We didn't support Values node yet because I wanted to think through some design decisions. This PR adds that support and the key decisions are called out below.

Values is NOT Part of Leaf Stage

We don't consider Values as a part of Leaf Stage, and values will have its worker assigned in either WorkerExchangeAssignmentRule, or the LiteModeWorkerssignmentRule.

This is also consistent with ServerPlanRequestVisitor.

Worker Assignment for Values

For now we will assign a random worker for the Values node. This logic is moved to PhysicalPlannerContext, since at present we store the encountered instances in there. Moreover, I am re-using the same method in Lite Mode worker assignment.

This also means that if a plan does not contain any table-scans, we will automatically assign the broker to all plan nodes, thereby executing the entire query in the broker. (no special handling required)

Solving Values Only Query in Broker

With v2 optimizer if a query plan does not consist of any table scan, we will solve it within broker itself.

Test Plan

• Added a bunch of plan based tests.
• Ran Quickstart and verified that queries work as expected (results pasted below)

Important

I realized that one of the tests in ResourceBasedQueriesTest is not yet enabled for the v2 optimizer. On enabling it I found that around 8 queries out of ~520 are failing, most of them are related to table partition hints.

Quickstart Tests

Ran the following queries:

SET useMultistageEngine=true;
SET usePhysicalOptimizer=true;

WITH tmp1(id, name) AS (
  VALUES
    (1, 'foo')
),
tmp2(user_id, nm2) AS (
    VALUES (1, 'bar')
  )
SELECT * FROM tmp1 JOIN tmp2 ON 1=1

SET useMultistageEngine=true;
SET usePhysicalOptimizer=true;

SELECT 1 as "timestamp"

SET useMultistageEngine=true;
SET usePhysicalOptimizer=true;

SELECT 1, 2

SET useMultistageEngine=true;
SET usePhysicalOptimizer=true;

SELECT COUNT(*) FROM userAttributes WHERE userUUID IS NULL

SET useMultistageEngine = true;
SET usePhysicalOptimizer = true;

SELECT COUNT(*)
FROM userAttributes
WHERE userUUID IN (
    SELECT userUUID
    FROM userGroups
    WHERE userUUID IS NULL
  )

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 94.44444% with 1 line in your changes missing coverage. Please review.

Project coverage is 63.25%. Comparing base (1a476de) to head (15349c0).
Report is 384 commits behind head on master.

Files with missing lines	Patch %	Lines
.../pinot/query/service/dispatch/QueryDispatcher.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #16221      +/-   ##
============================================
+ Coverage     62.90%   63.25%   +0.35%     
+ Complexity     1386     1362      -24     
============================================
  Files          2867     2962      +95     
  Lines        163354   171293    +7939     
  Branches      24952    26235    +1283     
============================================
+ Hits         102755   108355    +5600     
- Misses        52847    54730    +1883     
- Partials       7752     8208     +456     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-11	63.21% <94.44%> (+0.34%)	⬆️
java-21	63.19% <94.44%> (+0.36%)	⬆️
skip-bytebuffers-false	?
skip-bytebuffers-true	?
temurin	63.25% <94.44%> (+0.35%)	⬆️
unittests	63.25% <94.44%> (+0.35%)	⬆️
unittests1	64.76% <94.44%> (+8.94%)	⬆️
unittests2	33.42% <0.00%> (-0.15%)	⬇️

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.

[shauryachats]

@ankitsultana, if the Values node is being joined with a TableScan node, can we consider colocating the workers for the Values node to eliminate a shuffle?

[ankitsultana]

@ankitsultana, if the Values node is being joined with a TableScan node, can we consider colocating the workers for the Values node to eliminate a shuffle?

We can only do this if the scan was on a single node. If the scan is on multiple nodes, then we'll have to leverage "replicated execution" which is something I am working on to support lookup joins and supporting broadcast joins when a non dim-table has a complete copy available on all servers of the left side of the join.

When the scan is on a single node, even with the existing approach we'll pick that node for values which will lead to colocation.

[itschrispeck]

nit: avoid String.format

#14404

[itschrispeck]

could introduce some contention? maybe use ThreadLocalRandom?

[ankitsultana]

ah good call