This is an automated email from the ASF dual-hosted git repository.
deniskuzZ pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 89ea0b1b5ed HIVE-29262: [Addendum] Fix VectorPTFGroupBatches partition
key copy when partition and order columns overlap (#6597)
89ea0b1b5ed is described below
commit 89ea0b1b5edb8528c3be08588017216fbcc27beb
Author: Raghav Aggarwal <[email protected]>
AuthorDate: Sat Jul 11 19:58:36 2026 +0530
HIVE-29262: [Addendum] Fix VectorPTFGroupBatches partition key copy when
partition and order columns overlap (#6597)
---
.../ql/exec/vector/ptf/VectorPTFGroupBatches.java | 19 +-
.../hive/ql/exec/vector/ptf/VectorPTFOperator.java | 1 +
.../hive/ql/optimizer/physical/Vectorizer.java | 59 ------
.../exec/vector/ptf/TestVectorPTFGroupBatches.java | 1 +
.../vector_ptf_spill_partition_order_overlap.q | 39 ++++
.../vector_ptf_spill_partition_order_overlap.q.out | 226 +++++++++++++++++++++
6 files changed, 284 insertions(+), 61 deletions(-)
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
index f05683eb301..8fee32a569c 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFGroupBatches.java
@@ -26,6 +26,7 @@
import java.util.List;
import java.util.stream.Collectors;
+import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.common.type.HiveIntervalDayTime;
import org.apache.hadoop.hive.ql.exec.PTFPartition;
@@ -64,6 +65,7 @@ public class VectorPTFGroupBatches extends PTFPartition {
private int[] bufferedColumnMap;
private int[] orderColumnMap;
private int[] keyWithoutOrderColumnMap;
+ private int[] partitionKeyIndices;
private int spillLimitBufferedBatchCount;
private String spillLocalDirs;
@@ -289,13 +291,21 @@ public VectorPTFGroupBatches(Configuration hconf, int
vectorizedPTFMaxMemoryBuff
public void init(VectorPTFEvaluatorBase[] evaluators, int[]
outputProjectionColumnMap,
int[] bufferedColumnMap, TypeInfo[] bufferedTypeInfos, int[]
orderColumnMap,
- int[] keyWithoutOrderColumnMap, VectorizedRowBatch overflowBatch) {
+ int[] keyWithoutOrderColumnMap, int[] partitionColumnMap,
VectorizedRowBatch overflowBatch) {
this.evaluators = evaluators;
this.outputProjectionColumnMap = outputProjectionColumnMap;
this.bufferedColumnMap = bufferedColumnMap;
this.orderColumnMap = orderColumnMap;
this.keyWithoutOrderColumnMap = keyWithoutOrderColumnMap;
+ partitionKeyIndices = new int[keyWithoutOrderColumnMap.length];
+ for (int i = 0; i < keyWithoutOrderColumnMap.length; i++) {
+ partitionKeyIndices[i] = ArrayUtils.indexOf(partitionColumnMap,
keyWithoutOrderColumnMap[i]);
+ Preconditions.checkState(partitionKeyIndices[i] !=
ArrayUtils.INDEX_NOT_FOUND,
+ "Key input column %s not found among partition columns %s",
+ keyWithoutOrderColumnMap[i], Arrays.toString(partitionColumnMap));
+ }
+
this.overflowBatch = overflowBatch;
bufferedBatches = new ArrayList<BufferedVectorizedRowBatch>(0);
@@ -775,6 +785,11 @@ private void forwardSpilledBatches(VectorPTFOperator
vecPTFOperator, Object[] pa
}
}
+ /**
+ * Sets the partition key values as repeating columns in the overflowBatch.
Key columns are in
+ * output projection order, partitionKey values in PARTITION BY order;
partitionKeyIndices maps
+ * between the two.
+ */
private void copyPartitionColumnToOverflow(Object[] partitionKey) {
// Set partition column in overflowBatch.
// We can set by ref since our last batch is held by us.
@@ -782,7 +797,7 @@ private void copyPartitionColumnToOverflow(Object[]
partitionKey) {
for (int i = 0; i < keyInputColumnCount; i++) {
final int keyColumnNum = keyWithoutOrderColumnMap[i];
Preconditions.checkState(overflowBatch.cols[keyColumnNum] != null);
- setRepeatingColumn(partitionKey[i], overflowBatch, keyColumnNum);
+ setRepeatingColumn(partitionKey[partitionKeyIndices[i]], overflowBatch,
keyColumnNum);
}
}
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
index 6d03bb5ea5d..5accf6950c3 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/ptf/VectorPTFOperator.java
@@ -349,6 +349,7 @@ protected void initializeOp(Configuration hconf) throws
HiveException {
bufferedTypeInfos,
orderColumnMap,
keyWithoutOrderColumnMap,
+ vectorPTFInfo.getPartitionColumnMap(),
overflowBatch);
isFirstPartition = true;
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
index 8e041351ef3..82f81861a4d 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/Vectorizer.java
@@ -40,7 +40,6 @@
import java.util.TreeSet;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.StringUtils;
@@ -5077,59 +5076,6 @@ private static void createVectorPTFDesc(Operator<?
extends OperatorDesc> ptfOp,
vectorizedPTFMaxMemoryBufferingBatchCount);
}
- /**
- * Reorders partitionColumnMap and partitionColumnVectorTypes in-place so
that projected
- * partition-only columns come first in SELECT output order. Any partition
columns not found in
- * the output are appended at the end in their original plan order.
- */
- private static void
reorderPartitionColumnsToMatchOutputOrder(List<ColumnInfo> outputSignature,
- int evaluatorCount, int[] outputColumnProjectionMap, int[]
orderColumnMap,
- ExprNodeDesc[] partitionExprNodeDescs, int[] partitionColumnMap,
- Type[] partitionColumnVectorTypes) {
- final int count = partitionColumnMap.length;
- final int[] orderedMap = new int[count];
- final Type[] orderedTypes = new Type[count];
- final boolean[] placed = new boolean[count];
-
- int idx = 0;
- final int outputSize = outputSignature.size();
-
- for (int outputIdx = evaluatorCount; outputIdx < outputSize && idx <
count; outputIdx++) {
- final int outputColumn = outputColumnProjectionMap[outputIdx];
- final String colName = outputSignature.get(outputIdx).getInternalName();
- int matchedPartitionIdx = IntStream.range(0, count)
- .filter(p -> !placed[p])
- .filter(p -> partitionExprNodeDescs[p] instanceof ExprNodeColumnDesc
colDesc &&
- colDesc.getColumn().equals(colName))
- .findFirst()
- .orElse(-1);
-
- if (matchedPartitionIdx == -1) {
- matchedPartitionIdx = IntStream.range(0, count)
- .filter(p -> !placed[p] && partitionColumnMap[p] == outputColumn)
- .findFirst()
- .orElse(-1);
- }
-
- if (matchedPartitionIdx != -1 && !ArrayUtils.contains(orderColumnMap,
outputColumn)) {
- orderedMap[idx] = outputColumn;
- orderedTypes[idx] = partitionColumnVectorTypes[matchedPartitionIdx];
- placed[matchedPartitionIdx] = true;
- idx++;
- }
- }
-
- for (int p = 0; p < count; p++) {
- if (!placed[p]) {
- orderedMap[idx] = partitionColumnMap[p];
- orderedTypes[idx] = partitionColumnVectorTypes[p];
- idx++;
- }
- }
- System.arraycopy(orderedMap, 0, partitionColumnMap, 0, count);
- System.arraycopy(orderedTypes, 0, partitionColumnVectorTypes, 0, count);
- }
-
private static void determineKeyAndNonKeyInputColumnMap(int[]
outputColumnProjectionMap,
boolean isPartitionOrderBy, int[] orderColumnMap, int[]
partitionColumnMap,
int evaluatorCount, ArrayList<Integer> keyInputColumns,
@@ -5244,11 +5190,6 @@ private static VectorPTFInfo
createVectorPTFInfo(Operator<? extends OperatorDesc
int[] keyInputColumnMap =
ArrayUtils.toPrimitive(keyInputColumns.toArray(new Integer[0]));
int[] nonKeyInputColumnMap =
ArrayUtils.toPrimitive(nonKeyInputColumns.toArray(new Integer[0]));
- if (isPartitionOrderBy && partitionKeyCount > 1) {
- reorderPartitionColumnsToMatchOutputOrder(outputSignature,
evaluatorCount, outputColumnProjectionMap,
- orderColumnMap, partitionExprNodeDescs, partitionColumnMap,
partitionColumnVectorTypes);
- }
-
VectorExpression[][] evaluatorInputExpressions = new
VectorExpression[evaluatorCount][];
Type[][] evaluatorInputColumnVectorTypes = new Type[evaluatorCount][];
for (int i = 0; i < evaluatorCount; i++) {
diff --git
a/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/ptf/TestVectorPTFGroupBatches.java
b/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/ptf/TestVectorPTFGroupBatches.java
index 30395858177..c74d407c146 100644
---
a/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/ptf/TestVectorPTFGroupBatches.java
+++
b/ql/src/test/org/apache/hadoop/hive/ql/exec/vector/ptf/TestVectorPTFGroupBatches.java
@@ -588,6 +588,7 @@ private void init(VectorPTFGroupBatches groupBatches)
throws HiveException {
/* bufferedTypeInfos */ new TypeInfo[] { getTypeInfo("int"),
getTypeInfo("string") },
/* orderColumnMap */ new int[] { 1 }, // p_date
/* keyWithoutOrderColumnMap */ new int[] { 0 }, // p_mfgr
+ /* partitionColumnMap */ new int[] { 0 },
getFakeOperator().setupOverflowBatch(3, new String[] { "bigint",
"bigint" },
outputProjectionColumnMap, outputTypeInfos));
}
diff --git
a/ql/src/test/queries/clientpositive/vector_ptf_spill_partition_order_overlap.q
b/ql/src/test/queries/clientpositive/vector_ptf_spill_partition_order_overlap.q
new file mode 100644
index 00000000000..05de81530ea
--- /dev/null
+++
b/ql/src/test/queries/clientpositive/vector_ptf_spill_partition_order_overlap.q
@@ -0,0 +1,39 @@
+SET hive.vectorized.execution.enabled=true;
+SET hive.vectorized.execution.reduce.enabled=true;
+SET hive.vectorized.execution.ptf.enabled=true;
+SET hive.vectorized.testing.reducer.batch.size=2;
+SET hive.vectorized.ptf.max.memory.buffering.batch.count=1;
+
+SET hive.fetch.task.conversion=none;
+
+CREATE TABLE t1 (
+ dept STRING,
+ region BIGINT,
+ emp_id BIGINT,
+ salary DOUBLE
+) STORED AS ORC;
+
+INSERT INTO t1 values
+ ('engineering', 10, 1, 50000.0),
+ ('engineering', 10, 2, 55000.0),
+ ('engineering', 10, 3, 60000.0),
+ ('engineering', 10, 4, 45000.0),
+ ('engineering', 10, 5, 70000.0);
+
+EXPLAIN VECTORIZATION DETAIL
+SELECT dept, region, emp_id,
+ SUM(salary) OVER (
+ PARTITION BY dept, region
+ ORDER BY dept
+ ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ ) AS total
+FROM t1;
+
+SELECT dept, region, emp_id,
+ SUM(salary) OVER (
+ PARTITION BY dept, region
+ ORDER BY dept
+ ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ ) AS total
+FROM t1
+order by emp_id;
diff --git
a/ql/src/test/results/clientpositive/llap/vector_ptf_spill_partition_order_overlap.q.out
b/ql/src/test/results/clientpositive/llap/vector_ptf_spill_partition_order_overlap.q.out
new file mode 100644
index 00000000000..c785412cc56
--- /dev/null
+++
b/ql/src/test/results/clientpositive/llap/vector_ptf_spill_partition_order_overlap.q.out
@@ -0,0 +1,226 @@
+PREHOOK: query: CREATE TABLE t1 (
+ dept STRING,
+ region BIGINT,
+ emp_id BIGINT,
+ salary DOUBLE
+) STORED AS ORC
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@t1
+POSTHOOK: query: CREATE TABLE t1 (
+ dept STRING,
+ region BIGINT,
+ emp_id BIGINT,
+ salary DOUBLE
+) STORED AS ORC
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@t1
+PREHOOK: query: INSERT INTO t1 values
+ ('engineering', 10, 1, 50000.0),
+ ('engineering', 10, 2, 55000.0),
+ ('engineering', 10, 3, 60000.0),
+ ('engineering', 10, 4, 45000.0),
+ ('engineering', 10, 5, 70000.0)
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@t1
+POSTHOOK: query: INSERT INTO t1 values
+ ('engineering', 10, 1, 50000.0),
+ ('engineering', 10, 2, 55000.0),
+ ('engineering', 10, 3, 60000.0),
+ ('engineering', 10, 4, 45000.0),
+ ('engineering', 10, 5, 70000.0)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@t1
+POSTHOOK: Lineage: t1.dept SCRIPT []
+POSTHOOK: Lineage: t1.emp_id SCRIPT []
+POSTHOOK: Lineage: t1.region SCRIPT []
+POSTHOOK: Lineage: t1.salary SCRIPT []
+PREHOOK: query: EXPLAIN VECTORIZATION DETAIL
+SELECT dept, region, emp_id,
+ SUM(salary) OVER (
+ PARTITION BY dept, region
+ ORDER BY dept
+ ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ ) AS total
+FROM t1
+PREHOOK: type: QUERY
+PREHOOK: Input: default@t1
+#### A masked pattern was here ####
+POSTHOOK: query: EXPLAIN VECTORIZATION DETAIL
+SELECT dept, region, emp_id,
+ SUM(salary) OVER (
+ PARTITION BY dept, region
+ ORDER BY dept
+ ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ ) AS total
+FROM t1
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@t1
+#### A masked pattern was here ####
+PLAN VECTORIZATION:
+ enabled: true
+ enabledConditionsMet: [hive.vectorized.execution.enabled IS true]
+
+STAGE DEPENDENCIES:
+ Stage-1 is a root stage
+ Stage-0 depends on stages: Stage-1
+
+STAGE PLANS:
+ Stage: Stage-1
+ Tez
+#### A masked pattern was here ####
+ Edges:
+ Reducer 2 <- Map 1 (SIMPLE_EDGE)
+#### A masked pattern was here ####
+ Vertices:
+ Map 1
+ Map Operator Tree:
+ TableScan
+ alias: t1
+ Statistics: Num rows: 5 Data size: 595 Basic stats: COMPLETE
Column stats: COMPLETE
+ TableScan Vectorization:
+ native: true
+ vectorizationSchemaColumns: [0:dept:string,
1:region:bigint, 2:emp_id:bigint, 3:salary:double,
4:ROW__ID:struct<writeid:bigint,bucketid:int,rowid:bigint>,
5:ROW__IS__DELETED:boolean]
+ Reduce Output Operator
+ key expressions: dept (type: string), region (type: bigint)
+ null sort order: za
+ sort order: ++
+ Map-reduce partition columns: dept (type: string), region
(type: bigint)
+ Reduce Sink Vectorization:
+ className: VectorReduceSinkMultiKeyOperator
+ keyColumns: 0:string, 1:bigint
+ native: true
+ nativeConditionsMet:
hive.vectorized.execution.reducesink.new.enabled IS true, hive.execution.engine
tez IN [tez] IS true, No PTF TopN IS true, No DISTINCT columns IS true,
BinarySortableSerDe for keys IS true, LazyBinarySerDe for values IS true
+ valueColumns: 2:bigint, 3:double
+ Statistics: Num rows: 5 Data size: 595 Basic stats:
COMPLETE Column stats: COMPLETE
+ value expressions: emp_id (type: bigint), salary (type:
double)
+ Execution mode: vectorized, llap
+ LLAP IO: all inputs
+ Map Vectorization:
+ enabled: true
+ enabledConditionsMet:
hive.vectorized.use.vectorized.input.format IS true
+ inputFormatFeatureSupport: [DECIMAL_64]
+ featureSupportInUse: [DECIMAL_64]
+ inputFileFormats:
org.apache.hadoop.hive.ql.io.orc.OrcInputFormat
+ allNative: true
+ usesVectorUDFAdaptor: false
+ vectorized: true
+ rowBatchContext:
+ dataColumnCount: 4
+ includeColumns: [0, 1, 2, 3]
+ dataColumns: dept:string, region:bigint, emp_id:bigint,
salary:double
+ partitionColumnCount: 0
+ scratchColumnTypeNames: []
+ Reducer 2
+ Execution mode: vectorized, llap
+ Reduce Vectorization:
+ enabled: true
+ enableConditionsMet: hive.vectorized.execution.reduce.enabled
IS true, hive.execution.engine tez IN [tez] IS true
+ reduceColumnNullOrder: za
+ reduceColumnSortOrder: ++
+ allNative: false
+ usesVectorUDFAdaptor: false
+ vectorized: true
+ rowBatchContext:
+ dataColumnCount: 4
+ dataColumns: KEY.reducesinkkey0:string,
KEY.reducesinkkey1:bigint, VALUE._col0:bigint, VALUE._col1:double
+ partitionColumnCount: 0
+ scratchColumnTypeNames: [double]
+ Reduce Operator Tree:
+ Select Operator
+ expressions: KEY.reducesinkkey0 (type: string),
KEY.reducesinkkey1 (type: bigint), VALUE._col0 (type: bigint), VALUE._col1
(type: double)
+ outputColumnNames: _col0, _col1, _col2, _col3
+ Select Vectorization:
+ className: VectorSelectOperator
+ native: true
+ projectedOutputColumnNums: [0, 1, 2, 3]
+ Statistics: Num rows: 5 Data size: 595 Basic stats: COMPLETE
Column stats: COMPLETE
+ PTF Operator
+ Function definitions:
+ Input definition
+ input alias: ptf_0
+ output shape: _col0: string, _col1: bigint, _col2:
bigint, _col3: double
+ type: WINDOWING
+ Windowing table definition
+ input alias: ptf_1
+ name: windowingtablefunction
+ order by: _col0 ASC NULLS LAST
+ partition by: _col0, _col1
+ raw input shape:
+ window functions:
+ window function definition
+ alias: sum_window_0
+ arguments: _col3
+ name: sum
+ window function: GenericUDAFSumDouble
+ window frame: ROWS PRECEDING(MAX)~FOLLOWING(MAX)
+ PTF Vectorization:
+ allEvaluatorsAreStreaming: false
+ className: VectorPTFOperator
+ evaluatorClasses: [VectorPTFEvaluatorDoubleSum]
+ functionInputExpressions: [col 3:double]
+ functionNames: [sum]
+ keyInputColumns: [0, 1]
+ native: true
+ nonKeyInputColumns: [2, 3]
+ orderExpressions: [col 0:string]
+ outputColumns: [4, 0, 1, 2, 3]
+ outputTypes: [double, string, bigint, bigint, double]
+ partitionExpressions: [col 0:string, col 1:bigint]
+ streamingColumns: []
+ Statistics: Num rows: 5 Data size: 595 Basic stats: COMPLETE
Column stats: COMPLETE
+ Select Operator
+ expressions: _col0 (type: string), _col1 (type: bigint),
_col2 (type: bigint), sum_window_0 (type: double)
+ outputColumnNames: _col0, _col1, _col2, _col3
+ Select Vectorization:
+ className: VectorSelectOperator
+ native: true
+ projectedOutputColumnNums: [0, 1, 2, 4]
+ Statistics: Num rows: 5 Data size: 595 Basic stats:
COMPLETE Column stats: COMPLETE
+ File Output Operator
+ compressed: false
+ File Sink Vectorization:
+ className: VectorFileSinkOperator
+ native: false
+ Statistics: Num rows: 5 Data size: 595 Basic stats:
COMPLETE Column stats: COMPLETE
+ table:
+ input format:
org.apache.hadoop.mapred.SequenceFileInputFormat
+ output format:
org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat
+ serde:
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
+
+ Stage: Stage-0
+ Fetch Operator
+ limit: -1
+ Processor Tree:
+ ListSink
+
+PREHOOK: query: SELECT dept, region, emp_id,
+ SUM(salary) OVER (
+ PARTITION BY dept, region
+ ORDER BY dept
+ ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ ) AS total
+FROM t1
+order by emp_id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@t1
+#### A masked pattern was here ####
+POSTHOOK: query: SELECT dept, region, emp_id,
+ SUM(salary) OVER (
+ PARTITION BY dept, region
+ ORDER BY dept
+ ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ ) AS total
+FROM t1
+order by emp_id
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@t1
+#### A masked pattern was here ####
+engineering 10 1 280000.0
+engineering 10 2 280000.0
+engineering 10 3 280000.0
+engineering 10 4 280000.0
+engineering 10 5 280000.0
