Copilot commented on code in PR #6703:
URL: https://github.com/apache/hive/pull/6703#discussion_r3801637375
##########
ql/src/java/org/apache/hadoop/hive/ql/plan/PTFDesc.java:
##########
@@ -226,12 +226,7 @@ public String getStreamingColumns() {
public boolean getAllEvaluatorsAreStreaming() {
VectorPTFEvaluatorBase[] evaluators =
VectorPTFDesc.getEvaluators(vectorPTFDesc, vectorPTFInfo);
- for (VectorPTFEvaluatorBase evaluator : evaluators) {
- if (!evaluator.streamsResult()) {
- return false;
- }
- }
- return true;
+ return VectorPTFDesc.getAllEvaluatorsPurelyStreaming(evaluators);
}
Review Comment:
`getAllEvaluatorsAreStreaming()` now returns the result of
`getAllEvaluatorsPurelyStreaming(...)`, which is intentionally stricter (it
returns false for peer-group aggregated streaming evaluators like `cume_dist`).
This makes the method name misleading. Consider renaming the accessor (and any
related explain label) to reflect the new semantics (e.g.,
`getAllEvaluatorsArePurelyStreaming` / `isSinglePassStreaming`), or keep
`getAllEvaluatorsAreStreaming` but adjust the helper name/logic so the meaning
remains consistent.
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java:
##########
@@ -455,6 +476,22 @@ public void finishPartition() throws HiveException {
}
}
+ /**
+ * First pass for peer group aggregated streaming evaluators (e.g. cume_dist)
+ */
+ private void precomputeAggregatedStreamingResults() throws HiveException {
+ if (!isGroupAggregatedStreamingEvaluator) {
+ return;
+ }
+ for (VectorPTFEvaluatorBase evaluator : evaluators) {
+ if (evaluator.isGroupAggregatedStreamingEvaluator()) {
+ for (int groupRowCount : aggregatedGroupRowCounts) {
+ evaluator.addStreamingGroupResult(groupRowCount);
+ }
+ }
+ }
Review Comment:
`aggregatedGroupRowCounts` is only needed to precompute the per-peer-group
results; after this method completes, it’s no longer used for the current
partition. Clearing `aggregatedGroupRowCounts` here (after all evaluators have
consumed it) can reduce peak memory footprint for large partitions / many peer
groups.
##########
ql/src/test/queries/clientpositive/vector_ptf_cume_dist.q:
##########
@@ -0,0 +1,88 @@
+set hive.vectorized.testing.reducer.batch.size=2;
+
+CREATE TABLE vector_ptf_cume_dist_int(name string, rowindex int, mynumber int)
stored as orc;
Review Comment:
The test creates a fixed-name table without first dropping it. To avoid
failures when tests are re-run in the same metastore/session (or after partial
cleanup), add `DROP TABLE IF EXISTS vector_ptf_cume_dist_int;` before the
`CREATE TABLE` (and typically drop it at the end as well, consistent with other
qtests).
##########
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFEvaluatorCumeDist.java:
##########
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hadoop.hive.ql.exec.vector.ptf;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+
+import org.apache.hadoop.hive.ql.exec.vector.ColumnVector.Type;
+import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
+import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ptf.WindowFrameDef;
+
+/**
+ * This class evaluates cume_dist() for a PTF partition.
+ * Unlike rank(), cume_dist needs the total partition row count, group row
count, so it cannot produce a group's
+ * result while the group is still streaming in. It is therefore a peer group
aggregated streaming evaluator
+ * (see {@link VectorPTFEvaluatorBase#isGroupAggregatedStreamingEvaluator()}):
a first pass over the
+ * buffered group sizes precomputes each peer group's value via {@link
#addStreamingGroupResult(int)}
+ * (after {@link #setPartitionSize(int)} has been called), and the regular
streaming pass then just
+ * populates the precomputed values into the output column.
+ */
+public class VectorPTFEvaluatorCumeDist extends VectorPTFEvaluatorBase {
+
+ /**
+ * Per peer group cume_dist values computed in the first pass and consumed,
in order, by the
+ * streaming pass (one value is popped when a group's last batch is
processed).
+ */
+ private final Deque<Double> groupResults = new ArrayDeque<>();
+ private int rowPosition;
+
+ public VectorPTFEvaluatorCumeDist(WindowFrameDef windowFrameDef, int
outputColumnNum) {
+ super(windowFrameDef, outputColumnNum);
+ resetEvaluator();
+ }
+
+ @Override
+ public boolean needPartitionSize() {
+ return true;
+ }
+
+ @Override
+ public boolean isGroupAggregatedStreamingEvaluator() {
+ return true;
+ }
+
+ @Override
+ public void addStreamingGroupResult(int groupRowCount) throws HiveException {
+ if (partitionSize <= 0) {
+ throw new HiveException("Partition size must be set before precomputing
cume_dist");
+ }
+ rowPosition += groupRowCount;
+ groupResults.addLast(((double) rowPosition) / partitionSize);
+ }
+
+ @Override
+ public void evaluateGroupBatch(VectorizedRowBatch batch) throws
HiveException {
+ Double result = groupResults.peekFirst();
+ if (result == null) {
+ throw new HiveException("cume_dist streaming result is not available for
the current group");
+ }
+ DoubleColumnVector outputColVector = (DoubleColumnVector)
batch.cols[outputColumnNum];
+ outputColVector.isRepeating = true;
Review Comment:
The evaluator sets `isRepeating` and clears `isNull[0]`, but it does not
update `noNulls`. If `noNulls` was previously `false`, downstream operators may
treat the column as having nulls (at best causing extra null checks, at worst
interacting incorrectly with code that relies on `noNulls`). Set
`outputColVector.noNulls = true` when writing a non-null repeating value (and
keep `isNull[0] = false`).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
