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 f05683eb3018..8fee32a569c3 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(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 6d03bb5ea5d1..5accf6950c3a 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 8e041351ef35..82f81861a4dc 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 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 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 keyInputColumns, @@ -5244,11 +5190,6 @@ private static VectorPTFInfo createVectorPTFInfo(Operator 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 303958581779..c74d407c1462 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 000000000000..05de81530eab --- /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 000000000000..c785412cc562 --- /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, 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