You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by GitBox <gi...@apache.org> on 2021/06/08 02:10:55 UTC
[GitHub] [hive] maheshk114 opened a new pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
maheshk114 opened a new pull request #2364:
URL: https://github.com/apache/hive/pull/2364
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] github-actions[bot] commented on pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #2364:
URL: https://github.com/apache/hive/pull/2364#issuecomment-903016271
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.
Feel free to reach out on the dev@hive.apache.org list if the patch is in need of reviews.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #2364:
URL: https://github.com/apache/hive/pull/2364#discussion_r652474863
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
Review comment:
Referring to: https://issues.apache.org/jira/browse/HIVE-25108 maybe we could use a Hive exception instead?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
Review comment:
Was referring to: https://issues.apache.org/jira/browse/HIVE-25108 maybe we could use a Hive exception instead?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] github-actions[bot] closed pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #2364:
URL: https://github.com/apache/hive/pull/2364
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] github-actions[bot] closed pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #2364:
URL: https://github.com/apache/hive/pull/2364
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #2364:
URL: https://github.com/apache/hive/pull/2364#discussion_r652481213
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/util/FastByteComparisons.java
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.util;
+
+import com.google.common.primitives.UnsignedBytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.misc.Unsafe;
+
+import java.lang.reflect.Field;
+import java.nio.ByteOrder;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+abstract public class FastByteComparisons {
Review comment:
Was wondering why we dont import directly either from hadoop or tez?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] maheshk114 commented on a change in pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
maheshk114 commented on a change in pull request #2364:
URL: https://github.com/apache/hive/pull/2364#discussion_r652325810
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
Review comment:
done
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -1824,6 +1824,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
HIVEALIAS("hive.alias", "", ""),
HIVEMAPSIDEAGGREGATE("hive.map.aggr", true, "Whether to use map-side aggregation in Hive Group By queries"),
HIVEGROUPBYSKEW("hive.groupby.skewindata", false, "Whether there is skew in data to optimize group by queries"),
+
+ HIVE_ENABLE_COMBINER_FOR_GROUP_BY("hive.enable.combiner.for.groupby", true,
+ "Whether to enable tez combiner to aggregate the records after sorting is done. It is supported "
+ + "only for map side aggregation. This will be useful where the map side hash aggregation is converted to "
Review comment:
done
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
Review comment:
done
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -1824,6 +1824,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
HIVEALIAS("hive.alias", "", ""),
HIVEMAPSIDEAGGREGATE("hive.map.aggr", true, "Whether to use map-side aggregation in Hive Group By queries"),
HIVEGROUPBYSKEW("hive.groupby.skewindata", false, "Whether there is skew in data to optimize group by queries"),
+
+ HIVE_ENABLE_COMBINER_FOR_GROUP_BY("hive.enable.combiner.for.groupby", true,
Review comment:
done
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
Review comment:
Only few configs are copied to the tez tasks to avoid large data copy. These specific prefixes are used to filter out the configs which are passed to the executor tasks.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
Review comment:
The logging is done to retain the stack trace of the exception. This is useful while debugging.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
Review comment:
done
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
Review comment:
done
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ appendDirectlyToWriter(rawIter, writer);
+ } else {
+ processRows(rawIter, writer);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to combine rows", e);
Review comment:
The logging is done to retain the stack trace of the exception. This is useful while debugging.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ appendDirectlyToWriter(rawIter, writer);
+ } else {
+ processRows(rawIter, writer);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to combine rows", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ public static JobConf setCombinerInConf(BaseWork dest, JobConf conf, JobConf destConf) {
+ if (conf == null || !HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_ENABLE_COMBINER_FOR_GROUP_BY)) {
+ return conf;
+ }
+
+ // As of now, this function is called for all edges. We are interested only on edges
+ // to reducer.
+ if (!(dest instanceof ReduceWork)) {
+ return conf;
+ }
+ ReduceWork rw = (ReduceWork)dest;
+ Operator<?> reducer = rw.getReducer();
+ String combinerClass;
+
+ // In case of merge partial only we can have the combiner. In case of reduce side
+ // aggregation, the mode will be COMPLETE and reducer expects the records to be raw,
+ // not partially aggregated. For example, if its count, then reducer will not do a
+ // sum. Instead it will count the number of input records. This will result into
+ // invalid result if combiner has aggregated the records already.
+ if (reducer instanceof VectorGroupByOperator) {
+ VectorGroupByOperator operator = (VectorGroupByOperator) reducer;
+ if (operator.getConf().getMode() != GroupByDesc.Mode.MERGEPARTIAL) {
+ LOG.info("Combiner not set as operator is not MERGEPARTIAL" + reducer.toString());
+ return conf;
+ }
+ combinerClass = VectorGroupByCombiner.class.getName();
+ } else if (reducer instanceof GroupByOperator) {
+ GroupByOperator operator = (GroupByOperator) reducer;
+ if (operator.getConf().getMode() != GroupByDesc.Mode.MERGEPARTIAL) {
+ LOG.info("Combiner not set as operator is not MERGEPARTIAL" + reducer.toString());
+ return conf;
+ }
+ combinerClass = GroupByCombiner.class.getName();
+ } else {
+ LOG.info("Combiner not set as operator is not GroupByOperator" + reducer.toString());
+ return conf;
+ }
+
+ conf = new JobConf(conf);
+ String plan = HiveConf.getVar(destConf, HiveConf.ConfVars.PLAN);
+
+ // One source vertex can have multiple destination vertex. Add the destination vertex name
+ // to the config variable name to distinguish the plan of one reducer from other.
+ conf.set(getConfigPrefix(dest.getName()) + HiveConf.ConfVars.PLAN.varname, plan);
+ conf.set(getConfigPrefix(dest.getName()) + HAS_REDUCE_WORK, "true");
+
+ // Config with tez.runtime prefix are already part of included config.
+ conf.set("tez.runtime.combiner.class", combinerClass);
+
+ // This is adding the reduce work (destination node) plan path in the config for source map node.
+ // In the combiner we use the aggregation functions used at the reducer to aggregate
+ // the records. The mapper node (plan for map work) will not have these info so we will get
+ // this from reduce plan.
+ //TODO Need to check if we can get the aggregate info from map plan if its map side
Review comment:
removed the TODO as it can not be done.Incase of MAP side aggregate, the aggregation function used at MAP side and reduce side is different.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
Review comment:
The logging is done to retain the stack trace of the exception. This is useful while debugging.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
Review comment:
I am not sure ..many places the initBuffer for byte type is done. Anyways i will remove it and see.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ private void appendToWriter(DataInputBuffer val, DataInputBuffer key, IFile.Writer writer) {
+ try {
+ writer.append(key, val);
+ combineOutputRecordsCounter.increment(1);
+ } catch(IOException e) {
+ LOG.error("Append value list to writer failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ public static boolean equals(DataInputBuffer buf1, DataInputBuffer buf2) {
+ byte[] b1 = buf1.getData();
+ int s1 = buf1.getPosition();
+ int l1 = buf1.getLength() - buf1.getPosition();
+ byte[] b2 = buf2.getData();
+ int s2 = buf2.getPosition();
+ int l2 = buf2.getLength() - buf2.getPosition();
+ return org.apache.hadoop.hive.ql.exec.util.FastByteComparisons.equals(b1, s1, l1, b2, s2, l2);
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ //TODO this can be optimized further as RLE will still write length of value.
+ appendDirectlyToWriter(rawIter, writer);
+ return;
+ }
+
+ DataInputBuffer key = rawIter.getKey();
+ DataInputBuffer prevKey = new DataInputBuffer();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ DataInputBuffer valTemp = rawIter.getValue();
+ DataInputBuffer val = new DataInputBuffer();
+ val.reset(valTemp.getData(), valTemp.getPosition(),
+ valTemp.getLength() - valTemp.getPosition());
+
+ int numValues = 1;
+ long numRows = 1;
+
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!equals(prevKey, key)) {
+ if (numValues == 1) {
+ // if key has single record, no need for aggregation.
+ appendToWriter(val, prevKey, writer);
+ } else {
+ addValueToBatch(val, prevKey, writer, true);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numValues = 0;
+ } else {
+ addValueToBatch(val, prevKey, writer, false);
+ }
+ valTemp = rawIter.getValue();
+ val.reset(valTemp.getData(), valTemp.getPosition(),
+ valTemp.getLength() - valTemp.getPosition());
+ numRows++;
+ numValues++;
+ }
+
+ // Process the last key.
+ if (numValues == 1) {
+ appendToWriter(val, prevKey, writer);
+ } else {
+ addValueToBatch(val, prevKey, writer, true);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch (IOException | HiveException e) {
+ LOG.error("Failed to combine rows", e);
Review comment:
The logging is done to retain the stack trace of the exception. This is useful while debugging.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
Review comment:
where exactly this can be done ? "also: VectorizedBatchUtil.typeInfosFromStructObjectInspector(valueStructInspectors) could be reused below"
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ appendDirectlyToWriter(rawIter, writer);
+ } else {
+ processRows(rawIter, writer);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to combine rows", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ public static JobConf setCombinerInConf(BaseWork dest, JobConf conf, JobConf destConf) {
+ if (conf == null || !HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_ENABLE_COMBINER_FOR_GROUP_BY)) {
+ return conf;
+ }
+
+ // As of now, this function is called for all edges. We are interested only on edges
+ // to reducer.
+ if (!(dest instanceof ReduceWork)) {
Review comment:
I found the check bit non intuitive and added a comment. Thats why it's done separately.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/util/FastByteComparisons.java
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.util;
+
+import com.google.common.primitives.UnsignedBytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.misc.Unsafe;
+
+import java.lang.reflect.Field;
+import java.nio.ByteOrder;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+abstract public class FastByteComparisons {
Review comment:
copy of org.apache.tez.runtime.library.utils.FastByteComparisons
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
Review comment:
The logging is done to retain the stack trace of the exception. This is useful while debugging.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #2364:
URL: https://github.com/apache/hive/pull/2364#discussion_r652474863
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
Review comment:
Was referring to: https://issues.apache.org/jira/browse/HIVE-25141 maybe we could use a Hive exception instead?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #2364:
URL: https://github.com/apache/hive/pull/2364#discussion_r652474863
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
Review comment:
Referring to: https://issues.apache.org/jira/browse/HIVE-25108
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on a change in pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref commented on a change in pull request #2364:
URL: https://github.com/apache/hive/pull/2364#discussion_r648392171
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -1824,6 +1824,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
HIVEALIAS("hive.alias", "", ""),
HIVEMAPSIDEAGGREGATE("hive.map.aggr", true, "Whether to use map-side aggregation in Hive Group By queries"),
HIVEGROUPBYSKEW("hive.groupby.skewindata", false, "Whether there is skew in data to optimize group by queries"),
+
+ HIVE_ENABLE_COMBINER_FOR_GROUP_BY("hive.enable.combiner.for.groupby", true,
Review comment:
If its map specific maybe rename to HIVE_ENABLE_COMBINER_FOR_MAP_GROUP_BY (?)
##########
File path: common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
##########
@@ -1824,6 +1824,12 @@ private static void populateLlapDaemonVarsSet(Set<String> llapDaemonVarsSetLocal
HIVEALIAS("hive.alias", "", ""),
HIVEMAPSIDEAGGREGATE("hive.map.aggr", true, "Whether to use map-side aggregation in Hive Group By queries"),
HIVEGROUPBYSKEW("hive.groupby.skewindata", false, "Whether there is skew in data to optimize group by queries"),
+
+ HIVE_ENABLE_COMBINER_FOR_GROUP_BY("hive.enable.combiner.for.groupby", true,
+ "Whether to enable tez combiner to aggregate the records after sorting is done. It is supported "
+ + "only for map side aggregation. This will be useful where the map side hash aggregation is converted to "
Review comment:
nit. Mostly useful when map side hash aggregation switches to streaming mode due to high memory utilization etc.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ appendDirectlyToWriter(rawIter, writer);
+ } else {
+ processRows(rawIter, writer);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to combine rows", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ public static JobConf setCombinerInConf(BaseWork dest, JobConf conf, JobConf destConf) {
+ if (conf == null || !HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_ENABLE_COMBINER_FOR_GROUP_BY)) {
+ return conf;
+ }
+
+ // As of now, this function is called for all edges. We are interested only on edges
+ // to reducer.
+ if (!(dest instanceof ReduceWork)) {
Review comment:
nit. add this to the above condition
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ private void appendToWriter(DataInputBuffer val, DataInputBuffer key, IFile.Writer writer) {
+ try {
+ writer.append(key, val);
+ combineOutputRecordsCounter.increment(1);
+ } catch(IOException e) {
+ LOG.error("Append value list to writer failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ public static boolean equals(DataInputBuffer buf1, DataInputBuffer buf2) {
+ byte[] b1 = buf1.getData();
+ int s1 = buf1.getPosition();
+ int l1 = buf1.getLength() - buf1.getPosition();
+ byte[] b2 = buf2.getData();
+ int s2 = buf2.getPosition();
+ int l2 = buf2.getLength() - buf2.getPosition();
+ return org.apache.hadoop.hive.ql.exec.util.FastByteComparisons.equals(b1, s1, l1, b2, s2, l2);
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ //TODO this can be optimized further as RLE will still write length of value.
+ appendDirectlyToWriter(rawIter, writer);
+ return;
+ }
+
+ DataInputBuffer key = rawIter.getKey();
+ DataInputBuffer prevKey = new DataInputBuffer();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ DataInputBuffer valTemp = rawIter.getValue();
+ DataInputBuffer val = new DataInputBuffer();
+ val.reset(valTemp.getData(), valTemp.getPosition(),
+ valTemp.getLength() - valTemp.getPosition());
+
+ int numValues = 1;
+ long numRows = 1;
+
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!equals(prevKey, key)) {
+ if (numValues == 1) {
+ // if key has single record, no need for aggregation.
+ appendToWriter(val, prevKey, writer);
+ } else {
+ addValueToBatch(val, prevKey, writer, true);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numValues = 0;
+ } else {
+ addValueToBatch(val, prevKey, writer, false);
+ }
+ valTemp = rawIter.getValue();
+ val.reset(valTemp.getData(), valTemp.getPosition(),
+ valTemp.getLength() - valTemp.getPosition());
+ numRows++;
+ numValues++;
+ }
+
+ // Process the last key.
+ if (numValues == 1) {
+ appendToWriter(val, prevKey, writer);
+ } else {
+ addValueToBatch(val, prevKey, writer, true);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch (IOException | HiveException e) {
+ LOG.error("Failed to combine rows", e);
Review comment:
log or throw (not both)
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
Review comment:
Actually: createVectorizedRowBatch() already calls init as the end (throught reset) any particular reason we doing this again here?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
Review comment:
split this: to deserializeRead and vectorDeserializeRow
also: VectorizedBatchUtil.typeInfosFromStructObjectInspector(valueStructInspectors) could be reused below
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
+ }
+ }
+
+ batchValueDeserializer =
+ new VectorDeserializeRow<>(
+ new LazyBinaryDeserializeRead(
+ VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors),
+ true));
+ batchValueDeserializer.init(firstValueColumnOffset);
+
+ int[] valueColumnMap = new int[numValueCol];
+ for (int i = 0; i < numValueCol; i++) {
+ valueColumnMap[i] = i + firstValueColumnOffset;
+ }
+
+ valueLazyBinarySerializeWrite = new LazyBinarySerializeWrite(numValueCol);
+ valueVectorSerializeRow = new VectorSerializeRow<>(valueLazyBinarySerializeWrite);
+ valueVectorSerializeRow.init(VectorizedBatchUtil.typeInfosFromStructObjectInspector(
+ valueStructInspectors), valueColumnMap);
+ valueOutput = new ByteStream.Output();
+ valueVectorSerializeRow.setOutput(valueOutput);
+ outputBatch = batchContext.createVectorizedRowBatch();
+ valueBytesWritable = new DataInputBuffer();
+ }
+
+ private VectorAggregationBufferRow allocateAggregationBuffer() throws HiveException {
+ VectorAggregateExpression.AggregationBuffer[] aggregationBuffers =
+ new VectorAggregateExpression.AggregationBuffer[aggregators.length];
+ for (int i=0; i < aggregators.length; ++i) {
+ aggregationBuffers[i] = aggregators[i].getNewAggregationBuffer();
+ aggregators[i].reset(aggregationBuffers[i]);
+ }
+ return new VectorAggregationBufferRow(aggregationBuffers);
+ }
+
+ private void finishAggregation(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException, IOException {
+ for (int i = 0; i < aggregators.length; ++i) {
+ try {
+ aggregators[i].aggregateInput(aggregationBufferRow.getAggregationBuffer(i), inputBatch);
+ } catch (HiveException e) {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ // In case the input batch is full but the keys are still same we need not flush.
+ // Only evaluate the aggregates and store it in the aggregationBufferRow. The aggregate
+ // functions are incremental and will take care of correctness when next batch comes for
+ // aggregation.
+ if (!needFlush) {
+ return;
+ }
+
+ int colNum = firstValueColumnOffset;
+ for (int i = 0; i < aggregators.length; ++i) {
+ aggregators[i].assignRowColumn(outputBatch, 0, colNum++,
+ aggregationBufferRow.getAggregationBuffer(i));
+ }
+
+ valueLazyBinarySerializeWrite.reset();
+ valueVectorSerializeRow.serializeWrite(outputBatch, 0);
+ valueBytesWritable.reset(valueOutput.getData(), 0, valueOutput.getLength());
+ writer.append(key, valueBytesWritable);
+ combineOutputRecordsCounter.increment(1);
+ aggregationBufferRow.reset();
+ outputBatch.reset();
+ }
+
+ private void addValueToBatch(DataInputBuffer val, DataInputBuffer key,
+ IFile.Writer writer, boolean needFLush) throws IOException, HiveException {
+ batchValueDeserializer.setBytes(val.getData(), val.getPosition(),
+ val.getLength() - val.getPosition());
+ batchValueDeserializer.deserialize(inputBatch, inputBatch.size);
+ inputBatch.size++;
+ if (needFLush || (inputBatch.size >= VectorizedRowBatch.DEFAULT_SIZE)) {
+ processVectorGroup(key, writer, needFLush);
+ }
+ }
+
+ private void processVectorGroup(DataInputBuffer key, IFile.Writer writer, boolean needFlush)
+ throws HiveException {
+ try {
+ finishAggregation(key, writer, needFlush);
+ inputBatch.reset();
+ } catch (Exception e) {
+ String rowString;
+ try {
+ rowString = inputBatch.toString();
+ } catch (Exception e2) {
+ rowString = "[Error getting row data with exception "
+ + StringUtils.stringifyException(e2) + " ]";
+ }
+ LOG.error("Hive Runtime Error while processing vector batch" + rowString, e);
+ throw new HiveException("Hive Runtime Error while processing vector batch", e);
+ }
+ }
+
+ protected void appendDirectlyToWriter(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ do {
+ numRows++;
+ writer.append(rawIter.getKey(), rawIter.getValue());
+ } while (rawIter.next());
+ combineInputRecordsCounter.increment(numRows);
+ combineOutputRecordsCounter.increment(numRows);
+ } catch(IOException e) {
+ LOG.error("Append to writer failed", e);
Review comment:
Log or throw
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
Review comment:
getName() not needed -- and org.apache.hadoop.hive.ql.exec.GroupByCombiner -> GroupByCombiner
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
Review comment:
log or throw (not both)
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
Review comment:
make fields private and and final where possible
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
Review comment:
as above -- getName not needed
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ rw = null;
+ numValueCol = 0;
+ return;
+ }
+ }
+
+ try {
+ initObjectInspectors(rw.getTagToValueDesc().get(0), rw.getKeyDesc());
+ if (batchContext != null && numValueCol > 0) {
+ initVectorBatches();
+ }
+ } catch (SerDeException e) {
+ LOG.error("Fail to initialize VectorGroupByCombiner.", e);
+ throw new RuntimeException(e.getCause());
+ }
+ }
+
+ protected static String getConfigPrefix(String destName) {
+ return confPrefixForWorker + destName;
+ }
+
+ // Get the reduce work from the config. Here some hack is used to prefix the config name with
+ // "file." to avoid the config being filtered out.
+ private ReduceWork getReduceWork(TaskContext context) {
+ String destVertexName;
+ if (context instanceof TezOutputContextImpl) {
+ destVertexName = ((TezOutputContextImpl)context).getDestinationVertexName();
+ } else {
+ // As of now only map side combiner is supported.
+ return null;
+ }
+
+ String plan = conf.get(getConfigPrefix(destVertexName) +
+ HiveConf.ConfVars.PLAN.varname);
+ if (plan == null) {
+ LOG.info("Reduce plan is not set for vertex " + destVertexName);
+ return null;
+ }
+ this.conf.set(HiveConf.ConfVars.PLAN.varname, plan);
+ if (conf.getBoolean(getConfigPrefix(destVertexName)
+ + HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname,
+ true)) {
+ Path planPath = new Path(plan);
+ planPath = new Path(planPath, REDUCE_PLAN_NAME);
+ String planString = conf.get(getConfigPrefix(destVertexName) +
+ planPath.toUri().getPath());
+ this.conf.set(planPath.toUri().getPath(), planString);
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "true");
+ } else {
+ this.conf.set(HiveConf.ConfVars.HIVE_RPC_QUERY_PLAN.varname, "false");
+ }
+ this.conf.set(HAS_REDUCE_WORK, "true");
+ this.conf.set(MAPRED_REDUCER_CLASS, ExecReducer.class.getName());
+
+ return Utilities.getReduceWork(conf);
+ }
+
+ private void initObjectInspectors(TableDesc valueTableDesc,TableDesc keyTableDesc)
+ throws SerDeException {
+ inputKeyDeserializer =
+ ReflectionUtils.newInstance(keyTableDesc.getSerDeClass(), null);
+ inputKeyDeserializer.initialize(null, keyTableDesc.getProperties(), null);
+ keyObjectInspector = inputKeyDeserializer.getObjectInspector();
+
+ keyStructInspector = (StructObjectInspector) keyObjectInspector;
+ firstValueColumnOffset = keyStructInspector.getAllStructFieldRefs().size();
+
+ AbstractSerDe inputValueDeserializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ inputValueDeserializer.initialize(null, valueTableDesc.getProperties(), null);
+ valueObjectInspector = inputValueDeserializer.getObjectInspector();
+ valueStructInspectors = (StructObjectInspector) valueObjectInspector;
+ numValueCol = valueStructInspectors.getAllStructFieldRefs().size();
+ }
+
+ void initVectorBatches() throws HiveException {
+ inputBatch = batchContext.createVectorizedRowBatch();
+
+ // Create data buffers for value bytes column vectors.
+ for (int i = firstValueColumnOffset; i < inputBatch.numCols; i++) {
+ ColumnVector colVector = inputBatch.cols[i];
+ if (colVector instanceof BytesColumnVector) {
+ BytesColumnVector bytesColumnVector = (BytesColumnVector) colVector;
+ bytesColumnVector.initBuffer();
Review comment:
Believe init() is safer as other CV types might need some kind of initialisation too
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ appendDirectlyToWriter(rawIter, writer);
+ } else {
+ processRows(rawIter, writer);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to combine rows", e);
Review comment:
log or throw (not both)
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+ }
+
+ private void processAggregation(IFile.Writer writer, DataInputBuffer key)
+ throws Exception {
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ cachedValues[i] = aggregationEvaluators[i].evaluate(aggregationBuffers[i]);
+ }
+ BytesWritable result = (BytesWritable) valueSerializer.serialize(cachedValues,
+ aggrObjectInspector);
+ valueBuffer.reset(result.getBytes(), result.getLength());
+ writer.append(key, valueBuffer);
+ combineOutputRecordsCounter.increment(1);
+ for (int i = 0; i < aggregationEvaluators.length; i++) {
+ aggregationEvaluators[i].reset(aggregationBuffers[i]);
+ }
+ }
+
+ private void updateAggregation()
+ throws HiveException, SerDeException {
+ valWritable.set(prevVal.getData(), prevVal.getPosition(),
+ prevVal.getLength() - prevVal.getPosition());
+ Object row = valueSerializer.deserialize(valWritable);
+ groupByOperator.updateAggregation(row);
+ }
+
+ private void processRows(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ long numRows = 0;
+ try {
+ DataInputBuffer key = rawIter.getKey();
+ prevKey.reset(key.getData(), key.getPosition(), key.getLength() - key.getPosition());
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ int numSameKey = 0;
+ while (rawIter.next()) {
+ key = rawIter.getKey();
+ if (!VectorGroupByCombiner.equals(key, prevKey)) {
+ // if current key is not equal to the previous key then we have to emit the
+ // record. In case only one record was present for this key, then no need to
+ // do aggregation, We can directly append the key and value. For key with more
+ // than one record, we have to update the aggregation for the current value only
+ // as for previous values (records) aggregation is already done in previous
+ // iteration of loop.
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ prevKey.reset(key.getData(), key.getPosition(),
+ key.getLength() - key.getPosition());
+ numSameKey = 0;
+ } else {
+ // If there are more than one record with same key then update the aggregation.
+ updateAggregation();
+ numSameKey++;
+ }
+ prevVal.reset(rawIter.getValue().getData(), rawIter.getValue().getPosition(),
+ rawIter.getValue().getLength() - rawIter.getValue().getPosition());
+ numRows++;
+ }
+ if (numSameKey != 0) {
+ updateAggregation();
+ processAggregation(writer, prevKey);
+ } else {
+ writer.append(prevKey, prevVal);
+ }
+ combineInputRecordsCounter.increment(numRows);
+ } catch(Exception e) {
+ LOG.error("processRows failed", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void combine(TezRawKeyValueIterator rawIter, IFile.Writer writer) {
+ try {
+ if (!rawIter.next()) {
+ return;
+ }
+ if (numValueCol == 0) {
+ // For no aggregation, RLE in writer will take care of reduction.
+ appendDirectlyToWriter(rawIter, writer);
+ } else {
+ processRows(rawIter, writer);
+ }
+ } catch (IOException e) {
+ LOG.error("Failed to combine rows", e);
+ throw new RuntimeException(e.getMessage());
+ }
+ }
+
+ public static JobConf setCombinerInConf(BaseWork dest, JobConf conf, JobConf destConf) {
+ if (conf == null || !HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_ENABLE_COMBINER_FOR_GROUP_BY)) {
+ return conf;
+ }
+
+ // As of now, this function is called for all edges. We are interested only on edges
+ // to reducer.
+ if (!(dest instanceof ReduceWork)) {
+ return conf;
+ }
+ ReduceWork rw = (ReduceWork)dest;
+ Operator<?> reducer = rw.getReducer();
+ String combinerClass;
+
+ // In case of merge partial only we can have the combiner. In case of reduce side
+ // aggregation, the mode will be COMPLETE and reducer expects the records to be raw,
+ // not partially aggregated. For example, if its count, then reducer will not do a
+ // sum. Instead it will count the number of input records. This will result into
+ // invalid result if combiner has aggregated the records already.
+ if (reducer instanceof VectorGroupByOperator) {
+ VectorGroupByOperator operator = (VectorGroupByOperator) reducer;
+ if (operator.getConf().getMode() != GroupByDesc.Mode.MERGEPARTIAL) {
+ LOG.info("Combiner not set as operator is not MERGEPARTIAL" + reducer.toString());
+ return conf;
+ }
+ combinerClass = VectorGroupByCombiner.class.getName();
+ } else if (reducer instanceof GroupByOperator) {
+ GroupByOperator operator = (GroupByOperator) reducer;
+ if (operator.getConf().getMode() != GroupByDesc.Mode.MERGEPARTIAL) {
+ LOG.info("Combiner not set as operator is not MERGEPARTIAL" + reducer.toString());
+ return conf;
+ }
+ combinerClass = GroupByCombiner.class.getName();
+ } else {
+ LOG.info("Combiner not set as operator is not GroupByOperator" + reducer.toString());
+ return conf;
+ }
+
+ conf = new JobConf(conf);
+ String plan = HiveConf.getVar(destConf, HiveConf.ConfVars.PLAN);
+
+ // One source vertex can have multiple destination vertex. Add the destination vertex name
+ // to the config variable name to distinguish the plan of one reducer from other.
+ conf.set(getConfigPrefix(dest.getName()) + HiveConf.ConfVars.PLAN.varname, plan);
+ conf.set(getConfigPrefix(dest.getName()) + HAS_REDUCE_WORK, "true");
+
+ // Config with tez.runtime prefix are already part of included config.
+ conf.set("tez.runtime.combiner.class", combinerClass);
+
+ // This is adding the reduce work (destination node) plan path in the config for source map node.
+ // In the combiner we use the aggregation functions used at the reducer to aggregate
+ // the records. The mapper node (plan for map work) will not have these info so we will get
+ // this from reduce plan.
+ //TODO Need to check if we can get the aggregate info from map plan if its map side
Review comment:
ideally resolve TODO or replace with a ticket?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
Review comment:
not sure why we need a separate prefix for this -- the plan with the conf would already be on the workers right?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorGroupByCombiner.java
##########
@@ -0,0 +1,397 @@
+/*
+ * 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;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.exec.mr.ExecReducer;
+import org.apache.hadoop.hive.ql.exec.vector.expressions.aggregates.VectorAggregateExpression;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.ByteStream;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinaryDeserializeRead;
+import org.apache.hadoop.hive.serde2.lazybinary.fast.LazyBinarySerializeWrite;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.mapreduce.TaskCounter;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.tez.common.TezUtils;
+import org.apache.tez.common.counters.TezCounter;
+import org.apache.tez.mapreduce.combine.MRCombiner;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.api.impl.TezOutputContextImpl;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.IOException;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.MAPRED_REDUCER_CLASS;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for vectorized group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class VectorGroupByCombiner extends MRCombiner {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ VectorGroupByCombiner.class.getName());
+ protected final Configuration conf;
+ protected final TezCounter combineInputRecordsCounter;
+ protected final TezCounter combineOutputRecordsCounter;
+ VectorAggregateExpression[] aggregators;
+ VectorAggregationBufferRow aggregationBufferRow;
+ protected transient LazyBinarySerializeWrite valueLazyBinarySerializeWrite;
+
+ // This helper object serializes LazyBinary format reducer values from columns of a row
+ // in a vectorized row batch.
+ protected transient VectorSerializeRow<LazyBinarySerializeWrite> valueVectorSerializeRow;
+
+ // The output buffer used to serialize a value into.
+ protected transient ByteStream.Output valueOutput;
+ DataInputBuffer valueBytesWritable;
+
+ // Only required minimal configs are copied to the worker nodes. This hack (file.) is
+ // done to include these configs to be copied to the worker node.
+ protected static String confPrefixForWorker = "file.";
+
+ VectorDeserializeRow<LazyBinaryDeserializeRead> batchValueDeserializer;
+ int firstValueColumnOffset;
+ VectorizedRowBatchCtx batchContext = null;
+ protected int numValueCol = 0;
+ protected ReduceWork rw;
+ VectorizedRowBatch outputBatch = null;
+ VectorizedRowBatch inputBatch = null;
+ protected AbstractSerDe inputKeyDeserializer = null;
+ protected ObjectInspector keyObjectInspector = null;
+ protected ObjectInspector valueObjectInspector = null;
+ protected StructObjectInspector valueStructInspectors = null;
+ protected StructObjectInspector keyStructInspector = null;
+
+ public VectorGroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+
+ combineInputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_INPUT_RECORDS);
+ combineOutputRecordsCounter =
+ taskContext.getCounters().findCounter(TaskCounter.COMBINE_OUTPUT_RECORDS);
+
+ conf = TezUtils.createConfFromUserPayload(taskContext.getUserPayload());
+ rw = getReduceWork(taskContext);
+ if (rw == null) {
+ return;
+ }
+
+ if (rw.getReducer() instanceof VectorGroupByOperator) {
+ VectorGroupByOperator vectorGroupByOperator = (VectorGroupByOperator) rw.getReducer();
+ vectorGroupByOperator.initializeOp(this.conf);
+ this.aggregators = vectorGroupByOperator.getAggregators();
+ this.aggregationBufferRow = allocateAggregationBuffer();
+ batchContext = rw.getVectorizedRowBatchCtx();
+ if ((aggregators == null) || (aggregators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
Review comment:
Point to ticket
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/util/FastByteComparisons.java
##########
@@ -0,0 +1,256 @@
+/*
+ * 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.util;
+
+import com.google.common.primitives.UnsignedBytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import sun.misc.Unsafe;
+
+import java.lang.reflect.Field;
+import java.nio.ByteOrder;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+
+abstract public class FastByteComparisons {
Review comment:
is this a copy of org.apache.hadoop.io.FastByteComparisons ?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/exec/GroupByCombiner.java
##########
@@ -0,0 +1,275 @@
+/*
+ * 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;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByCombiner;
+import org.apache.hadoop.hive.ql.exec.vector.VectorGroupByOperator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.plan.BaseWork;
+import org.apache.hadoop.hive.ql.plan.GroupByDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceWork;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
+import org.apache.hadoop.hive.serde2.AbstractSerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.io.BytesWritable;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.util.ReflectionUtils;
+import org.apache.tez.runtime.api.TaskContext;
+import org.apache.tez.runtime.library.common.sort.impl.IFile;
+import org.apache.tez.runtime.library.common.sort.impl.TezRawKeyValueIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.hive.ql.exec.Utilities.HAS_REDUCE_WORK;
+import static org.apache.hadoop.hive.ql.exec.Utilities.REDUCE_PLAN_NAME;
+
+// Combiner for normal group by operator. In case of map side aggregate, the partially
+// aggregated records are sorted based on group by key. If because of some reasons, like hash
+// table memory exceeded the limit or the first few batches of records have less ndvs, the
+// aggregation is not done, then here the aggregation can be done cheaply as the records
+// are sorted based on group by key.
+public class GroupByCombiner extends VectorGroupByCombiner {
+
+ private static final Logger LOG = LoggerFactory.getLogger(
+ org.apache.hadoop.hive.ql.exec.GroupByCombiner.class.getName());
+
+ private transient GenericUDAFEvaluator[] aggregationEvaluators;
+ AbstractSerDe valueSerializer;
+ GenericUDAFEvaluator.AggregationBuffer[] aggregationBuffers;
+ GroupByOperator groupByOperator;
+ ObjectInspector aggrObjectInspector;
+ DataInputBuffer valueBuffer;
+ Object[] cachedValues;
+ DataInputBuffer prevKey;
+ BytesWritable valWritable;
+ DataInputBuffer prevVal;
+
+ public GroupByCombiner(TaskContext taskContext) throws HiveException, IOException {
+ super(taskContext);
+ if (rw != null) {
+ try {
+ groupByOperator = (GroupByOperator) rw.getReducer();
+
+ ArrayList<ObjectInspector> ois = new ArrayList<ObjectInspector>();
+ ois.add(keyObjectInspector);
+ ois.add(valueObjectInspector);
+ ObjectInspector[] rowObjectInspector = new ObjectInspector[1];
+ rowObjectInspector[0] =
+ ObjectInspectorFactory.getStandardStructObjectInspector(Utilities.reduceFieldNameList,
+ ois);
+ groupByOperator.setInputObjInspectors(rowObjectInspector);
+ groupByOperator.initializeOp(conf);
+ aggregationBuffers = groupByOperator.getAggregationBuffers();
+ aggregationEvaluators = groupByOperator.getAggregationEvaluator();
+
+ TableDesc valueTableDesc = rw.getTagToValueDesc().get(0);
+ if ((aggregationEvaluators == null) || (aggregationEvaluators.length != numValueCol)) {
+ //TODO : Need to support distinct. The logic has to be changed to extract only
+ // those aggregates which are not part of distinct.
+ LOG.info(" Combiner is disabled as the number of value columns does" +
+ " not match with number of aggregators");
+ numValueCol = 0;
+ rw = null;
+ return;
+ }
+ valueSerializer = (AbstractSerDe) ReflectionUtils.newInstance(
+ valueTableDesc.getSerDeClass(), null);
+ valueSerializer.initialize(null,
+ valueTableDesc.getProperties(), null);
+
+ aggrObjectInspector = groupByOperator.getAggrObjInspector();
+ valueBuffer = new DataInputBuffer();
+ cachedValues = new Object[aggregationEvaluators.length];
+ prevKey = new DataInputBuffer();
+ valWritable = new BytesWritable();
+ prevVal = new DataInputBuffer();
+ } catch (Exception e) {
+ LOG.error(" GroupByCombiner failed", e);
Review comment:
log or throw
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref commented on pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref commented on pull request #2364:
URL: https://github.com/apache/hive/pull/2364#issuecomment-903886213
Shall we revive this @maheshk114 ? Seems like most of the comments have already been addressed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] pgaref removed a comment on pull request #2364: Hive 24471 : Add support for combiner in hash mode group aggregation
Posted by GitBox <gi...@apache.org>.
pgaref removed a comment on pull request #2364:
URL: https://github.com/apache/hive/pull/2364#issuecomment-903886213
Shall we revive this @maheshk114 ? Seems like most of the comments have already been addressed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org