HIVE-25124: PTF: Vectorize cume_dist function

[ramitg254]

Change-Id: I4b03ac73b116d9e61f482745a2ff1f8f6ad1ab77

What changes were proposed in this pull request?

added support for vectorised cumulative distribution ptf

Why are the changes needed?

currently cume_dist() evaluates only in non vectorized manner so added evaluator for vectorized as well

Does this PR introduce any user-facing change?

No

How was this patch tested?

added q test as well as ci results

[Copilot]

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

This PR adds vectorized PTF support for cume_dist() in Hive, including a new vector evaluator and updated LLAP golden outputs plus new query-based coverage.

Changes:

• Add VectorPTFEvaluatorCumeDist and wire it into VectorPTFDesc/Vectorizer supported-function handling.
• Extend the vectorized PTF execution path to handle peer-group aggregated streaming (needed by cume_dist) and partition-size dependency.
• Add a new positive QTest and update multiple .q.out baselines to reflect newly vectorized plans.

Reviewed changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
ql/src/java/org/apache/hadoop/hive/ql/plan/VectorPTFDesc.java	Registers cume_dist as a supported vectorized PTF function and adds helper to detect “purely streaming” evaluator sets.
ql/src/java/org/apache/hadoop/hive/ql/plan/PTFDesc.java	Reuses new “purely streaming” detection when reporting streaming capability.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java	Allows cume_dist through validation path similar to rank/dense_rank.
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java	Adjusts operator control flow to use “purely streaming” vs buffered partition finishing.
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java	Adds peer-group row-count aggregation + partition-size propagation to support cume_dist precomputation.
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorBase.java	Adds partition-size plumbing and APIs for peer-group aggregated streaming evaluators.
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorCumeDist.java	Implements vectorized cume_dist using a two-pass approach (precompute per peer group, then stream results).
ql/src/test/queries/clientpositive/vector_ptf_cume_dist.q	New QTest covering vectorized and non-vectorized cume_dist behavior + vectorization explain.
ql/src/test/results/clientpositive/llap/vector_ptf_cume_dist.q.out	Golden output for the new vector_ptf_cume_dist.q test.
ql/src/test/results/clientpositive/llap/vectorized_ptf.q.out	Updates baseline explain output due to new vectorization behavior/reasons.
ql/src/test/results/clientpositive/llap/vector_windowing.q.out	Updates baseline to reflect cume_dist now being supported in vectorized PTF.
ql/src/test/results/clientpositive/llap/vector_windowing_rank.q.out	Updates baseline to reflect cume_dist now vectorizing in the tested plan.
ql/src/test/results/clientpositive/llap/vector_windowing_gby2.q.out	Updates baseline supported-function list due to cume_dist addition.
ql/src/test/queries/clientpositive/sketches_rewrite_cume_dist.q	Temporarily disables vectorized PTF for a query with complex-type passthrough columns.
ql/src/test/queries/clientpositive/sketches_rewrite_cume_dist_partition_by.q	Same as above for the partition-by variant.
ql/src/test/results/clientpositive/llap/sketches_materialized_view_cume_dist.q.out	Baseline changes due to altered vectorization/execution mode after the feature change.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[abstractdog]

thanks @ramitg254 so far, left some comments

[abstractdog]

currently needPartitionSize is only true when isGroupAggregatedStreamingEvaluator is true, what about completely removing needPartitionSize check and interface method?

[ramitg254]

they were added because I want to separate out them:
needPartitionSize : solely intended for partition size (would be helpful when adding vectorized support for percent rank).
isGroupAggregatedStreamingEvaluator: solely intended for group aggregation of row count currently but can be extended to any kind of other group wise aggregation if needed later on.

please check this : #6703 (comment)

[ramitg254]

kept needPartitionSize due to this reasoning #6703 (comment) and extracted this long if check to a separate helper method isPurelyStreamingEvaluator
wdyt?

[abstractdog]

the more I look at this code the more I feel this evaluator is not a streaming one :( from the point it needs to have group counts, and makes getAllEvaluatorsAreStreaming return false, I believe we have no reason the to treat it like a streaming one, because it doesn't make the code fall into optimized codepaths anyway, like

    if (allEvaluatorsAreStreaming) {
      // We can process this batch immediately.
      groupBatches.evaluateStreamingGroupBatch(batch, isLastGroupBatch);
      vectorForward(batch);

if you agree, please edit the javadoc for the VectorPTFEvaluatorCumeDist class too, and similify code introduced by this PR

[ramitg254]

yes It is not purely streaming, but does streams result in the outputcolumn vector for every row based on the precomputation we are doing rather than the aggregation functions like avg, sum etc. which just compute their results over data columns and not concerned with output column vector:

hive/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java

Line 443 in c56c3f5

if (!evaluator.streamsResult()) {

so we need to bypass that and directly populate the results in

hive/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java

Line 639 in c56c3f5

groupBatches.fillGroupResultsAndForward(this, partitionKey);

So in general we are does streaming results but we can't use that optimization you mentioned because it is not possible to do it on the fly and rather we need to buffer batches so that we can have parition size and other precomputations.
I think we can keep it as true, wdyt?

[abstractdog]

okay, I'm fine with that, please find place as javadoc for the corresponding reasoning why cume_dist is special

1. Output is per-row, not per-partition. Different peer groups get different values. So it can't be modeled as a buffered aggregate (sum, avg, first_value) which produces one value broadcast to the partition.
2. The value can't be known while the group is still streaming in. The denominator needs partition_size; the numerator needs the group's end position. So it can't be modeled as a pure per-row streamer (row_number, rank, dense_rank) either.

It's the first vectorized function that sits in a third category: per-group output, computed after the partition is fully seen. percent_rank and a strict ntile will land in the same category when they get vectorized.

Why the current shape (streamsResult() == true + isGroupAggregatedStreamingEvaluator() == true) is right:

The two questions the operator actually needs answered are orthogonal:

• Does output flow into the column vector per batch? → yes (streamsResult)
• Is a pre-pass over peer-group counts required before that flow starts? → yes (isGroupAggregatedStreamingEvaluator)

please take extra care in documenting the VectorPTFDesc functions:

getAllEvaluatorsAreStreaming
isPurelyStreamingEvaluator

feel free to remove isPurelyStreamingEvaluator, it doesn't add value, a proper javadoc on getAllEvaluatorsAreStreaming would be better

also given this condition:

   return evaluator.streamsResult()
        && !evaluator.isGroupAggregatedStreamingEvaluator()
        && !evaluator.needPartitionSize();

is needPartitionSize really needed? I think, for this special case, we need 2 separate booleans, and isGroupAggregatedStreamingEvaluator handles the new scenario properly

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
1.0% Duplication on New Code

See analysis details on SonarQube Cloud