HIVE-11303: Getting Tez LimitExceededException after dag execution on large query (Jason Dere, reviewed by Vikram Dixit)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/04d54f61 Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/04d54f61 Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/04d54f61 Branch: refs/heads/master Commit: 04d54f61c9f56906160936751e772080c079498c Parents: 9904162 Author: Jason Dere <jd...@hortonworks.com> Authored: Tue Jul 21 14:03:12 2015 -0700 Committer: Jason Dere <jd...@hortonworks.com> Committed: Tue Jul 21 14:03:12 2015 -0700 ---------------------------------------------------------------------- .../org/apache/hadoop/hive/conf/HiveConf.java | 3 + .../test/resources/testconfiguration.properties | 4 + .../apache/hadoop/hive/ql/exec/JoinUtil.java | 87 +- .../hadoop/hive/ql/exec/MapJoinOperator.java | 2 +- .../apache/hadoop/hive/ql/exec/Operator.java | 6 + .../hive/ql/exec/tez/KeyValuesAdapter.java | 47 ++ .../hive/ql/exec/tez/KeyValuesFromKeyValue.java | 90 +++ .../ql/exec/tez/KeyValuesFromKeyValues.java | 48 ++ .../hive/ql/exec/tez/ReduceRecordProcessor.java | 11 +- .../hive/ql/exec/tez/ReduceRecordSource.java | 12 +- .../ql/exec/vector/VectorMapJoinOperator.java | 1 - .../mapjoin/VectorMapJoinCommonOperator.java | 1 + .../hive/ql/optimizer/ConvertJoinMapJoin.java | 213 +++-- .../hive/ql/optimizer/MapJoinProcessor.java | 44 +- .../ql/optimizer/ReduceSinkMapJoinProc.java | 84 +- .../hadoop/hive/ql/parse/GenTezProcContext.java | 12 + .../hadoop/hive/ql/parse/GenTezUtils.java | 23 +- .../apache/hadoop/hive/ql/parse/GenTezWork.java | 81 +- .../apache/hadoop/hive/ql/plan/BaseWork.java | 2 +- .../hive/ql/plan/CommonMergeJoinDesc.java | 4 + .../hadoop/hive/ql/plan/ExprNodeDescUtils.java | 115 +++ .../apache/hadoop/hive/ql/plan/MapJoinDesc.java | 11 + .../apache/hadoop/hive/ql/plan/ReduceWork.java | 2 +- .../clientpositive/tez_dynpart_hashjoin_1.q | 101 +++ .../clientpositive/tez_dynpart_hashjoin_2.q | 83 ++ .../tez_vector_dynpart_hashjoin_1.q | 102 +++ .../tez_vector_dynpart_hashjoin_2.q | 84 ++ .../tez/tez_dynpart_hashjoin_1.q.out | 791 ++++++++++++++++++ .../tez/tez_dynpart_hashjoin_2.q.out | 564 +++++++++++++ .../tez/tez_vector_dynpart_hashjoin_1.q.out | 804 +++++++++++++++++++ .../tez/tez_vector_dynpart_hashjoin_2.q.out | 570 +++++++++++++ 31 files changed, 3899 insertions(+), 103 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java ---------------------------------------------------------------------- diff --git a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java index 39477d6..33b67dd 100644 --- a/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java +++ b/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java @@ -1046,6 +1046,9 @@ public class HiveConf extends Configuration { "job, process those skewed keys. The same key need not be skewed for all the tables, and so,\n" + "the follow-up map-reduce job (for the skewed keys) would be much faster, since it would be a\n" + "map-join."), + HIVEDYNAMICPARTITIONHASHJOIN("hive.optimize.dynamic.partition.hashjoin", false, + "Whether to enable dynamically partitioned hash join optimization. \n" + + "This setting is also dependent on enabling hive.auto.convert.join"), HIVECONVERTJOIN("hive.auto.convert.join", true, "Whether Hive enables the optimization about converting common join into mapjoin based on the input file size"), HIVECONVERTJOINNOCONDITIONALTASK("hive.auto.convert.join.noconditionaltask", true, http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/itests/src/test/resources/testconfiguration.properties ---------------------------------------------------------------------- diff --git a/itests/src/test/resources/testconfiguration.properties b/itests/src/test/resources/testconfiguration.properties index 97715fc..fbde465 100644 --- a/itests/src/test/resources/testconfiguration.properties +++ b/itests/src/test/resources/testconfiguration.properties @@ -334,6 +334,10 @@ minitez.query.files=bucket_map_join_tez1.q,\ tez_dml.q,\ tez_fsstat.q,\ tez_insert_overwrite_local_directory_1.q,\ + tez_dynpart_hashjoin_1.q,\ + tez_dynpart_hashjoin_2.q,\ + tez_vector_dynpart_hashjoin_1.q,\ + tez_vector_dynpart_hashjoin_2.q,\ tez_join_hash.q,\ tez_join_result_complex.q,\ tez_join_tests.q,\ http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/JoinUtil.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/JoinUtil.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/JoinUtil.java index 7b57550..0aaa51a 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/JoinUtil.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/JoinUtil.java @@ -40,6 +40,7 @@ import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils; import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption; import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector; +import org.apache.hadoop.hive.serde2.objectinspector.StructField; import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector; import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; import org.apache.hadoop.mapred.Reporter; @@ -65,8 +66,21 @@ public class JoinUtil { int iterate = Math.min(exprEntries.length, inputObjInspector.length); for (byte alias = 0; alias < iterate; alias++) { + ObjectInspector inputOI = inputObjInspector[alias]; + + // For vectorized reduce-side operators getting inputs from a reduce sink, + // the row object inspector will get a flattened version of the object inspector + // where the nested key/value structs are replaced with a single struct: + // Example: { key: { reducesinkkey0:int }, value: { _col0:int, _col1:int, .. } } + // Would get converted to the following for a vectorized input: + // { 'key.reducesinkkey0':int, 'value._col0':int, 'value._col1':int, .. } + // The ExprNodeEvaluator initialzation below gets broken with the flattened + // object inpsectors, so convert it back to the a form that contains the + // nested key/value structs. + inputOI = unflattenObjInspector(inputOI); + if (alias == (byte) posBigTableAlias || - exprEntries[alias] == null || inputObjInspector[alias] == null) { + exprEntries[alias] == null || inputOI == null) { // skip the driver and directly loadable tables continue; } @@ -74,7 +88,7 @@ public class JoinUtil { List<ExprNodeEvaluator> exprList = exprEntries[alias]; List<ObjectInspector> fieldOIList = new ArrayList<ObjectInspector>(); for (int i = 0; i < exprList.size(); i++) { - fieldOIList.add(exprList.get(i).initialize(inputObjInspector[alias])); + fieldOIList.add(exprList.get(i).initialize(inputOI)); } result[alias] = fieldOIList; } @@ -350,4 +364,73 @@ public class JoinUtil { rc.setTableDesc(tblDesc); return rc; } + + private static String KEY_FIELD_PREFIX = (Utilities.ReduceField.KEY + ".").toLowerCase(); + private static String VALUE_FIELD_PREFIX = (Utilities.ReduceField.VALUE + ".").toLowerCase(); + + /** + * Create a new struct object inspector for the list of struct fields, first removing the + * prefix from the field name. + * @param fields + * @param prefixToRemove + * @return + */ + private static ObjectInspector createStructFromFields(List<StructField> fields, String prefixToRemove) { + int prefixLength = prefixToRemove.length() + 1; // also remove the '.' after the prefix + ArrayList<String> fieldNames = new ArrayList<String>(); + ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>(); + for (StructField field : fields) { + fieldNames.add(field.getFieldName().substring(prefixLength)); + fieldOIs.add(field.getFieldObjectInspector()); + } + return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs); + } + + /** + * Checks the input object inspector to see if it is in for form of a flattened struct + * like the ones generated by a vectorized reduce sink input: + * { 'key.reducesinkkey0':int, 'value._col0':int, 'value._col1':int, .. } + * If so, then it creates an "unflattened" struct that contains nested key/value + * structs: + * { key: { reducesinkkey0:int }, value: { _col0:int, _col1:int, .. } } + * + * @param oi + * @return unflattened object inspector if unflattening is needed, + * otherwise the original object inspector + */ + private static ObjectInspector unflattenObjInspector(ObjectInspector oi) { + if (oi instanceof StructObjectInspector) { + // Check if all fields start with "key." or "value." + // If so, then unflatten by adding an additional level of nested key and value structs + // Example: { "key.reducesinkkey0":int, "key.reducesinkkey1": int, "value._col6":int } + // Becomes + // { "key": { "reducesinkkey0":int, "reducesinkkey1":int }, "value": { "_col6":int } } + ArrayList<StructField> keyFields = new ArrayList<StructField>(); + ArrayList<StructField> valueFields = new ArrayList<StructField>(); + for (StructField field : ((StructObjectInspector) oi).getAllStructFieldRefs()) { + String fieldNameLower = field.getFieldName().toLowerCase(); + if (fieldNameLower.startsWith(KEY_FIELD_PREFIX)) { + keyFields.add(field); + } else if (fieldNameLower.startsWith(VALUE_FIELD_PREFIX)) { + valueFields.add(field); + } else { + // Not a flattened struct, no need to unflatten + return oi; + } + } + + // All field names are of the form "key." or "value." + // Create key/value structs and add the respective fields to each one + ArrayList<ObjectInspector> reduceFieldOIs = new ArrayList<ObjectInspector>(); + reduceFieldOIs.add(createStructFromFields(keyFields, Utilities.ReduceField.KEY.toString())); + reduceFieldOIs.add(createStructFromFields(valueFields, Utilities.ReduceField.VALUE.toString())); + + // Finally create the outer struct to contain the key, value structs + return ObjectInspectorFactory.getStandardStructObjectInspector( + Utilities.reduceFieldNameList, + reduceFieldOIs); + } + + return oi; + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java index a40f0a9..1b9d7ef 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/MapJoinOperator.java @@ -154,7 +154,7 @@ public class MapJoinOperator extends AbstractMapJoinOperator<MapJoinDesc> implem final ExecMapperContext mapContext = getExecContext(); final MapredContext mrContext = MapredContext.get(); - if (!conf.isBucketMapJoin()) { + if (!conf.isBucketMapJoin() && !conf.isDynamicPartitionHashJoin()) { /* * The issue with caching in case of bucket map join is that different tasks * process different buckets and if the container is reused to join a different bucket, http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java index d7f1b42..0f02737 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java @@ -1354,4 +1354,10 @@ public abstract class Operator<T extends OperatorDesc> implements Serializable,C return childOperators; } } + + public void removeParents() { + for (Operator<?> parent : new ArrayList<Operator<?>>(getParentOperators())) { + removeParent(parent); + } + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesAdapter.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesAdapter.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesAdapter.java new file mode 100644 index 0000000..8f706fe --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesAdapter.java @@ -0,0 +1,47 @@ +/** + * 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.tez; + +import java.io.IOException; + +/** + * Key-values interface for the Reader used by ReduceRecordSource + */ +public interface KeyValuesAdapter { + /** + * Get the key for current record + * @return + * @throws IOException + */ + Object getCurrentKey() throws IOException; + + /** + * Get the values for the current record + * @return + * @throws IOException + */ + Iterable<Object> getCurrentValues() throws IOException; + + /** + * Move to the next record + * @return true if successful, false if there are no more records to process + * @throws IOException + */ + boolean next() throws IOException; +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValue.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValue.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValue.java new file mode 100644 index 0000000..51cdeca --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValue.java @@ -0,0 +1,90 @@ +/** + * 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.tez; + +import java.io.IOException; +import java.util.Iterator; +import java.util.NoSuchElementException; +import org.apache.tez.runtime.library.api.KeyValueReader; + +/** + * Provides a key/values (note the plural values) interface out of a KeyValueReader, + * needed by ReduceRecordSource when reading input from a key/value source. + */ +public class KeyValuesFromKeyValue implements KeyValuesAdapter { + protected KeyValueReader reader; + protected ValueIterator<Object> valueIterator = + new ValueIterator<Object>(); + + private static class ValueIterator<T> implements Iterator<T>, Iterable<T> { + + protected boolean hasNextValue = false; + protected T value = null; + + @Override + public boolean hasNext() { + return hasNextValue; + } + + @Override + public T next() { + if (!hasNextValue) { + throw new NoSuchElementException(); + } + hasNextValue = false; + return value; + } + + void reset(T value) { + this.value = value; + hasNextValue = true; + } + + @Override + public void remove() { + throw new UnsupportedOperationException(); + } + + @Override + public Iterator<T> iterator() { + return this; + } + } + + public KeyValuesFromKeyValue(KeyValueReader reader) { + this.reader = reader; + } + + @Override + public Object getCurrentKey() throws IOException { + return reader.getCurrentKey(); + } + + @Override + public Iterable<Object> getCurrentValues() throws IOException { + Object obj = reader.getCurrentValue(); + valueIterator.reset(obj); + return valueIterator; + } + + @Override + public boolean next() throws IOException { + return reader.next(); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValues.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValues.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValues.java new file mode 100644 index 0000000..b027bce --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/KeyValuesFromKeyValues.java @@ -0,0 +1,48 @@ +/** + * 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.tez; + +import java.io.IOException; +import org.apache.tez.runtime.library.api.KeyValuesReader; + +/** + * Provides a key/values interface out of a KeyValuesReader for use by ReduceRecordSource. + */ +public class KeyValuesFromKeyValues implements KeyValuesAdapter { + protected KeyValuesReader reader; + + public KeyValuesFromKeyValues(KeyValuesReader reader) { + this.reader = reader; + } + + @Override + public Object getCurrentKey() throws IOException { + return reader.getCurrentKey(); + } + + @Override + public Iterable<Object> getCurrentValues() throws IOException { + return reader.getCurrentValues(); + } + + @Override + public boolean next() throws IOException { + return reader.next(); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java index 545d7c6..d649672 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java @@ -50,6 +50,7 @@ import org.apache.tez.runtime.api.Input; import org.apache.tez.runtime.api.LogicalInput; import org.apache.tez.runtime.api.LogicalOutput; import org.apache.tez.runtime.api.ProcessorContext; +import org.apache.tez.runtime.api.Reader; import org.apache.tez.runtime.library.api.KeyValuesReader; /** @@ -129,10 +130,11 @@ public class ReduceRecordProcessor extends RecordProcessor{ tagToReducerMap.put(mergeReduceWork.getTag(), mergeReduceWork); } - bigTablePosition = (byte) reduceWork.getTag(); ((TezContext) MapredContext.get()).setDummyOpsMap(connectOps); } + bigTablePosition = (byte) reduceWork.getTag(); + ObjectInspector[] mainWorkOIs = null; ((TezContext) MapredContext.get()).setInputs(inputs); ((TezContext) MapredContext.get()).setTezProcessorContext(processorContext); @@ -227,10 +229,13 @@ public class ReduceRecordProcessor extends RecordProcessor{ reducer.setParentOperators(null); // clear out any parents as reducer is the root TableDesc keyTableDesc = redWork.getKeyDesc(); - KeyValuesReader reader = (KeyValuesReader) inputs.get(inputName).getReader(); + Reader reader = inputs.get(inputName).getReader(); sources[tag] = new ReduceRecordSource(); - sources[tag].init(jconf, redWork.getReducer(), redWork.getVectorMode(), keyTableDesc, + // Only the big table input source should be vectorized (if applicable) + // Note this behavior may have to change if we ever implement a vectorized merge join + boolean vectorizedRecordSource = (tag == bigTablePosition) && redWork.getVectorMode(); + sources[tag].init(jconf, redWork.getReducer(), vectorizedRecordSource, keyTableDesc, valueTableDesc, reader, tag == bigTablePosition, (byte) tag, redWork.getVectorScratchColumnTypeMap()); ois[tag] = sources[tag].getObjectInspector(); http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java index 20f6dba..89f7572 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordSource.java @@ -57,6 +57,8 @@ import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.util.ReflectionUtils; import org.apache.hadoop.util.StringUtils; +import org.apache.tez.runtime.api.Reader; +import org.apache.tez.runtime.library.api.KeyValueReader; import org.apache.tez.runtime.library.api.KeyValuesReader; /** @@ -107,7 +109,7 @@ public class ReduceRecordSource implements RecordSource { /* this is only used in the error code path */ private List<VectorExpressionWriter> valueStringWriters; - private KeyValuesReader reader; + private KeyValuesAdapter reader; private boolean handleGroupKey; @@ -120,7 +122,7 @@ public class ReduceRecordSource implements RecordSource { private final GroupIterator groupIterator = new GroupIterator(); void init(JobConf jconf, Operator<?> reducer, boolean vectorized, TableDesc keyTableDesc, - TableDesc valueTableDesc, KeyValuesReader reader, boolean handleGroupKey, byte tag, + TableDesc valueTableDesc, Reader reader, boolean handleGroupKey, byte tag, Map<Integer, String> vectorScratchColumnTypeMap) throws Exception { @@ -129,7 +131,11 @@ public class ReduceRecordSource implements RecordSource { this.reducer = reducer; this.vectorized = vectorized; this.keyTableDesc = keyTableDesc; - this.reader = reader; + if (reader instanceof KeyValueReader) { + this.reader = new KeyValuesFromKeyValue((KeyValueReader) reader); + } else { + this.reader = new KeyValuesFromKeyValues((KeyValuesReader) reader); + } this.handleGroupKey = handleGroupKey; this.tag = tag; http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java index e9bd44a..9bd811c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorMapJoinOperator.java @@ -100,7 +100,6 @@ public class VectorMapJoinOperator extends VectorMapJoinBaseOperator { @Override public Collection<Future<?>> initializeOp(Configuration hconf) throws HiveException { - // Use a final variable to properly parameterize the processVectorInspector closure. // Using a member variable in the closure will not do the right thing... final int parameterizePosBigTable = conf.getPosBigTable(); http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java index 4c8c4b1..87ebcf2 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/VectorMapJoinCommonOperator.java @@ -551,6 +551,7 @@ public abstract class VectorMapJoinCommonOperator extends MapJoinOperator implem @Override protected Collection<Future<?>> initializeOp(Configuration hconf) throws HiveException { + Collection<Future<?>> result = super.initializeOp(hconf); if (LOG.isDebugEnabled()) { http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java index 5a87bd6..e3acdfc 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ConvertJoinMapJoin.java @@ -93,9 +93,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { if (retval == null) { return retval; } else { - int pos = 0; // it doesn't matter which position we use in this case. - convertJoinSMBJoin(joinOp, context, pos, 0, false); - return null; + fallbackToReduceSideJoin(joinOp, context); } } @@ -103,27 +101,8 @@ public class ConvertJoinMapJoin implements NodeProcessor { // exact number of buckets. Else choose the largest number of estimated // reducers from the parent operators. int numBuckets = -1; - int estimatedBuckets = -1; if (context.conf.getBoolVar(HiveConf.ConfVars.HIVE_CONVERT_JOIN_BUCKET_MAPJOIN_TEZ)) { - for (Operator<? extends OperatorDesc>parentOp : joinOp.getParentOperators()) { - if (parentOp.getOpTraits().getNumBuckets() > 0) { - numBuckets = (numBuckets < parentOp.getOpTraits().getNumBuckets()) ? - parentOp.getOpTraits().getNumBuckets() : numBuckets; - } - - if (parentOp instanceof ReduceSinkOperator) { - ReduceSinkOperator rs = (ReduceSinkOperator) parentOp; - estimatedBuckets = (estimatedBuckets < rs.getConf().getNumReducers()) ? - rs.getConf().getNumReducers() : estimatedBuckets; - } - } - - if (numBuckets <= 0) { - numBuckets = estimatedBuckets; - if (numBuckets <= 0) { - numBuckets = 1; - } - } + numBuckets = estimateNumBuckets(joinOp, true); } else { numBuckets = 1; } @@ -136,7 +115,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { } else { // only case is full outer join with SMB enabled which is not possible. Convert to regular // join. - convertJoinSMBJoin(joinOp, context, 0, 0, false); + fallbackToReduceSideJoin(joinOp, context); return null; } } @@ -155,20 +134,18 @@ public class ConvertJoinMapJoin implements NodeProcessor { if (mapJoinConversionPos < 0) { // we are just converting to a common merge join operator. The shuffle // join in map-reduce case. - int pos = 0; // it doesn't matter which position we use in this case. - convertJoinSMBJoin(joinOp, context, pos, 0, false); + fallbackToReduceSideJoin(joinOp, context); return null; } - MapJoinOperator mapJoinOp = convertJoinMapJoin(joinOp, context, mapJoinConversionPos); + MapJoinOperator mapJoinOp = convertJoinMapJoin(joinOp, context, mapJoinConversionPos, true); // map join operator by default has no bucket cols and num of reduce sinks // reduced by 1 - mapJoinOp -.setOpTraits(new OpTraits(null, -1, null)); + mapJoinOp.setOpTraits(new OpTraits(null, -1, null)); mapJoinOp.setStatistics(joinOp.getStatistics()); // propagate this change till the next RS for (Operator<? extends OperatorDesc> childOp : mapJoinOp.getChildOperators()) { - setAllChildrenTraitsToNull(childOp); + setAllChildrenTraits(childOp, mapJoinOp.getOpTraits()); } return null; @@ -180,7 +157,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { // we cannot convert to bucket map join, we cannot convert to // map join either based on the size. Check if we can convert to SMB join. if (context.conf.getBoolVar(HiveConf.ConfVars.HIVE_AUTO_SORTMERGE_JOIN) == false) { - convertJoinSMBJoin(joinOp, context, 0, 0, false); + fallbackToReduceSideJoin(joinOp, context); return null; } Class<? extends BigTableSelectorForAutoSMJ> bigTableMatcherClass = null; @@ -209,8 +186,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { // contains aliases from sub-query // we are just converting to a common merge join operator. The shuffle // join in map-reduce case. - int pos = 0; // it doesn't matter which position we use in this case. - convertJoinSMBJoin(joinOp, context, pos, 0, false); + fallbackToReduceSideJoin(joinOp, context); return null; } @@ -220,8 +196,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { } else { // we are just converting to a common merge join operator. The shuffle // join in map-reduce case. - int pos = 0; // it doesn't matter which position we use in this case. - convertJoinSMBJoin(joinOp, context, pos, 0, false); + fallbackToReduceSideJoin(joinOp, context); } return null; } @@ -317,16 +292,16 @@ public class ConvertJoinMapJoin implements NodeProcessor { mergeJoinOp.cloneOriginalParentsList(mergeJoinOp.getParentOperators()); } - private void setAllChildrenTraitsToNull(Operator<? extends OperatorDesc> currentOp) { + private void setAllChildrenTraits(Operator<? extends OperatorDesc> currentOp, OpTraits opTraits) { if (currentOp instanceof ReduceSinkOperator) { return; } - currentOp.setOpTraits(new OpTraits(null, -1, null)); + currentOp.setOpTraits(new OpTraits(opTraits.getBucketColNames(), opTraits.getNumBuckets(), opTraits.getSortCols())); for (Operator<? extends OperatorDesc> childOp : currentOp.getChildOperators()) { if ((childOp instanceof ReduceSinkOperator) || (childOp instanceof GroupByOperator)) { break; } - setAllChildrenTraitsToNull(childOp); + setAllChildrenTraits(childOp, opTraits); } } @@ -338,7 +313,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { return false; } - MapJoinOperator mapJoinOp = convertJoinMapJoin(joinOp, context, bigTablePosition); + MapJoinOperator mapJoinOp = convertJoinMapJoin(joinOp, context, bigTablePosition, true); MapJoinDesc joinDesc = mapJoinOp.getConf(); joinDesc.setBucketMapJoin(true); @@ -633,7 +608,7 @@ public class ConvertJoinMapJoin implements NodeProcessor { */ public MapJoinOperator convertJoinMapJoin(JoinOperator joinOp, OptimizeTezProcContext context, - int bigTablePosition) throws SemanticException { + int bigTablePosition, boolean removeReduceSink) throws SemanticException { // bail on mux operator because currently the mux operator masks the emit keys // of the constituent reduce sinks. for (Operator<? extends OperatorDesc> parentOp : joinOp.getParentOperators()) { @@ -646,45 +621,49 @@ public class ConvertJoinMapJoin implements NodeProcessor { MapJoinOperator mapJoinOp = MapJoinProcessor.convertJoinOpMapJoinOp(context.conf, joinOp, joinOp.getConf().isLeftInputJoin(), joinOp.getConf().getBaseSrc(), - joinOp.getConf().getMapAliases(), bigTablePosition, true); + joinOp.getConf().getMapAliases(), bigTablePosition, true, removeReduceSink); mapJoinOp.getConf().setHybridHashJoin(HiveConf.getBoolVar(context.conf, - HiveConf.ConfVars.HIVEUSEHYBRIDGRACEHASHJOIN)); + HiveConf.ConfVars.HIVEUSEHYBRIDGRACEHASHJOIN)); Operator<? extends OperatorDesc> parentBigTableOp = mapJoinOp.getParentOperators().get(bigTablePosition); if (parentBigTableOp instanceof ReduceSinkOperator) { - for (Operator<?> p : parentBigTableOp.getParentOperators()) { - // we might have generated a dynamic partition operator chain. Since - // we're removing the reduce sink we need do remove that too. - Set<Operator<?>> dynamicPartitionOperators = new HashSet<Operator<?>>(); - Map<Operator<?>, AppMasterEventOperator> opEventPairs = new HashMap<>(); - for (Operator<?> c : p.getChildOperators()) { - AppMasterEventOperator event = findDynamicPartitionBroadcast(c); - if (event != null) { - dynamicPartitionOperators.add(c); - opEventPairs.put(c, event); + if (removeReduceSink) { + for (Operator<?> p : parentBigTableOp.getParentOperators()) { + // we might have generated a dynamic partition operator chain. Since + // we're removing the reduce sink we need do remove that too. + Set<Operator<?>> dynamicPartitionOperators = new HashSet<Operator<?>>(); + Map<Operator<?>, AppMasterEventOperator> opEventPairs = new HashMap<>(); + for (Operator<?> c : p.getChildOperators()) { + AppMasterEventOperator event = findDynamicPartitionBroadcast(c); + if (event != null) { + dynamicPartitionOperators.add(c); + opEventPairs.put(c, event); + } } - } - for (Operator<?> c : dynamicPartitionOperators) { - if (context.pruningOpsRemovedByPriorOpt.isEmpty() || - !context.pruningOpsRemovedByPriorOpt.contains(opEventPairs.get(c))) { - p.removeChild(c); - // at this point we've found the fork in the op pipeline that has the pruning as a child plan. - LOG.info("Disabling dynamic pruning for: " - + ((DynamicPruningEventDesc) opEventPairs.get(c).getConf()).getTableScan().getName() - + ". Need to be removed together with reduce sink"); + for (Operator<?> c : dynamicPartitionOperators) { + if (context.pruningOpsRemovedByPriorOpt.isEmpty() || + !context.pruningOpsRemovedByPriorOpt.contains(opEventPairs.get(c))) { + p.removeChild(c); + // at this point we've found the fork in the op pipeline that has the pruning as a child plan. + LOG.info("Disabling dynamic pruning for: " + + ((DynamicPruningEventDesc) opEventPairs.get(c).getConf()).getTableScan().getName() + + ". Need to be removed together with reduce sink"); + } + } + for (Operator<?> op : dynamicPartitionOperators) { + context.pruningOpsRemovedByPriorOpt.add(opEventPairs.get(op)); } } - for (Operator<?> op : dynamicPartitionOperators) { - context.pruningOpsRemovedByPriorOpt.add(opEventPairs.get(op)); + + mapJoinOp.getParentOperators().remove(bigTablePosition); + if (!(mapJoinOp.getParentOperators().contains(parentBigTableOp.getParentOperators().get(0)))) { + mapJoinOp.getParentOperators().add(bigTablePosition, + parentBigTableOp.getParentOperators().get(0)); } + parentBigTableOp.getParentOperators().get(0).removeChild(parentBigTableOp); } - mapJoinOp.getParentOperators().remove(bigTablePosition); - if (!(mapJoinOp.getParentOperators().contains(parentBigTableOp.getParentOperators().get(0)))) { - mapJoinOp.getParentOperators().add(bigTablePosition, - parentBigTableOp.getParentOperators().get(0)); - } - parentBigTableOp.getParentOperators().get(0).removeChild(parentBigTableOp); + for (Operator<? extends OperatorDesc>op : mapJoinOp.getParentOperators()) { if (!(op.getChildOperators().contains(mapJoinOp))) { op.getChildOperators().add(mapJoinOp); @@ -720,4 +699,100 @@ public class ConvertJoinMapJoin implements NodeProcessor { return null; } + + /** + * Estimate the number of buckets in the join, using the parent operators' OpTraits and/or + * parent operators' number of reducers + * @param joinOp + * @param useOpTraits Whether OpTraits should be used for the estimate. + * @return + */ + private static int estimateNumBuckets(JoinOperator joinOp, boolean useOpTraits) { + int numBuckets = -1; + int estimatedBuckets = -1; + + for (Operator<? extends OperatorDesc>parentOp : joinOp.getParentOperators()) { + if (parentOp.getOpTraits().getNumBuckets() > 0) { + numBuckets = (numBuckets < parentOp.getOpTraits().getNumBuckets()) ? + parentOp.getOpTraits().getNumBuckets() : numBuckets; + } + + if (parentOp instanceof ReduceSinkOperator) { + ReduceSinkOperator rs = (ReduceSinkOperator) parentOp; + estimatedBuckets = (estimatedBuckets < rs.getConf().getNumReducers()) ? + rs.getConf().getNumReducers() : estimatedBuckets; + } + } + + if (!useOpTraits) { + // Ignore the value we got from OpTraits. + // The logic below will fall back to the estimate from numReducers + numBuckets = -1; + } + + if (numBuckets <= 0) { + numBuckets = estimatedBuckets; + if (numBuckets <= 0) { + numBuckets = 1; + } + } + + return numBuckets; + } + + private boolean convertJoinDynamicPartitionedHashJoin(JoinOperator joinOp, OptimizeTezProcContext context) + throws SemanticException { + // Attempt dynamic partitioned hash join + // Since we don't have big table index yet, must start with estimate of numReducers + int numReducers = estimateNumBuckets(joinOp, false); + LOG.info("Try dynamic partitioned hash join with estimated " + numReducers + " reducers"); + int bigTablePos = getMapJoinConversionPos(joinOp, context, numReducers); + if (bigTablePos >= 0) { + // Now that we have the big table index, get real numReducers value based on big table RS + ReduceSinkOperator bigTableParentRS = + (ReduceSinkOperator) (joinOp.getParentOperators().get(bigTablePos)); + numReducers = bigTableParentRS.getConf().getNumReducers(); + LOG.debug("Real big table reducers = " + numReducers); + + MapJoinOperator mapJoinOp = convertJoinMapJoin(joinOp, context, bigTablePos, false); + if (mapJoinOp != null) { + LOG.info("Selected dynamic partitioned hash join"); + mapJoinOp.getConf().setDynamicPartitionHashJoin(true); + // Set OpTraits for dynamically partitioned hash join: + // bucketColNames: Re-use previous joinOp's bucketColNames. Parent operators should be + // reduce sink, which should have bucket columns based on the join keys. + // numBuckets: set to number of reducers + // sortCols: This is an unsorted join - no sort cols + OpTraits opTraits = new OpTraits( + joinOp.getOpTraits().getBucketColNames(), + numReducers, + null); + mapJoinOp.setOpTraits(opTraits); + mapJoinOp.setStatistics(joinOp.getStatistics()); + // propagate this change till the next RS + for (Operator<? extends OperatorDesc> childOp : mapJoinOp.getChildOperators()) { + setAllChildrenTraits(childOp, mapJoinOp.getOpTraits()); + } + return true; + } + } + + return false; + } + + private void fallbackToReduceSideJoin(JoinOperator joinOp, OptimizeTezProcContext context) + throws SemanticException { + if (context.conf.getBoolVar(HiveConf.ConfVars.HIVECONVERTJOIN) && + context.conf.getBoolVar(HiveConf.ConfVars.HIVEDYNAMICPARTITIONHASHJOIN)) { + if (convertJoinDynamicPartitionedHashJoin(joinOp, context)) { + return; + } + } + + // we are just converting to a common merge join operator. The shuffle + // join in map-reduce case. + int pos = 0; // it doesn't matter which position we use in this case. + LOG.info("Fallback to common merge join operator"); + convertJoinSMBJoin(joinOp, context, pos, 0, false); + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java index 4d84f0f..f8f2b7b 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/MapJoinProcessor.java @@ -29,6 +29,8 @@ import java.util.Map; import java.util.Set; import java.util.Stack; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.common.ObjectPair; import org.apache.hadoop.hive.conf.HiveConf; @@ -57,6 +59,7 @@ import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx; import org.apache.hadoop.hive.ql.lib.Rule; import org.apache.hadoop.hive.ql.lib.RuleRegExp; import org.apache.hadoop.hive.ql.parse.GenMapRedWalker; +import org.apache.hadoop.hive.ql.parse.OptimizeTezProcContext; import org.apache.hadoop.hive.ql.parse.ParseContext; import org.apache.hadoop.hive.ql.parse.SemanticException; import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; @@ -89,6 +92,7 @@ public class MapJoinProcessor implements Transform { // (column type + column name). The column name is not really used anywhere, but it // needs to be passed. Use the string defined below for that. private static final String MAPJOINKEY_FIELDPREFIX = "mapjoinkey"; + private static final Log LOG = LogFactory.getLog(MapJoinProcessor.class.getName()); public MapJoinProcessor() { } @@ -356,11 +360,18 @@ public class MapJoinProcessor implements Transform { public static MapJoinOperator convertJoinOpMapJoinOp(HiveConf hconf, JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases, int mapJoinPos, boolean noCheckOuterJoin) throws SemanticException { + return convertJoinOpMapJoinOp(hconf, op, leftInputJoin, baseSrc, mapAliases, + mapJoinPos, noCheckOuterJoin, true); + } + + public static MapJoinOperator convertJoinOpMapJoinOp(HiveConf hconf, + JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases, + int mapJoinPos, boolean noCheckOuterJoin, boolean adjustParentsChildren) + throws SemanticException { MapJoinDesc mapJoinDescriptor = getMapJoinDesc(hconf, op, leftInputJoin, baseSrc, mapAliases, - mapJoinPos, noCheckOuterJoin); - + mapJoinPos, noCheckOuterJoin, adjustParentsChildren); // reduce sink row resolver used to generate map join op RowSchema outputRS = op.getSchema(); @@ -1025,7 +1036,7 @@ public class MapJoinProcessor implements Transform { public static MapJoinDesc getMapJoinDesc(HiveConf hconf, JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases, - int mapJoinPos, boolean noCheckOuterJoin) throws SemanticException { + int mapJoinPos, boolean noCheckOuterJoin, boolean adjustParentsChildren) throws SemanticException { JoinDesc desc = op.getConf(); JoinCondDesc[] condns = desc.getConds(); Byte[] tagOrder = desc.getTagOrder(); @@ -1072,6 +1083,26 @@ public class MapJoinProcessor implements Transform { // get the join keys from old parent ReduceSink operators Map<Byte, List<ExprNodeDesc>> keyExprMap = pair.getSecond(); + if (!adjustParentsChildren) { + // Since we did not remove reduce sink parents, keep the original value expressions + newValueExprs = valueExprs; + + // Join key exprs are represented in terms of the original table columns, + // we need to convert these to the generated column names we can see in the Join operator + Map<Byte, List<ExprNodeDesc>> newKeyExprMap = new HashMap<Byte, List<ExprNodeDesc>>(); + for (Map.Entry<Byte, List<ExprNodeDesc>> mapEntry : keyExprMap.entrySet()) { + Byte pos = mapEntry.getKey(); + ReduceSinkOperator rsParent = oldReduceSinkParentOps.get(pos.byteValue()); + List<ExprNodeDesc> keyExprList = + ExprNodeDescUtils.resolveJoinKeysAsRSColumns(mapEntry.getValue(), rsParent); + if (keyExprList == null) { + throw new SemanticException("Error resolving join keys"); + } + newKeyExprMap.put(pos, keyExprList); + } + keyExprMap = newKeyExprMap; + } + // construct valueTableDescs and valueFilteredTableDescs List<TableDesc> valueTableDescs = new ArrayList<TableDesc>(); List<TableDesc> valueFilteredTableDescs = new ArrayList<TableDesc>(); @@ -1163,4 +1194,11 @@ public class MapJoinProcessor implements Transform { return mapJoinDescriptor; } + + public static MapJoinDesc getMapJoinDesc(HiveConf hconf, + JoinOperator op, boolean leftInputJoin, String[] baseSrc, List<String> mapAliases, + int mapJoinPos, boolean noCheckOuterJoin) throws SemanticException { + return getMapJoinDesc(hconf, op, leftInputJoin, baseSrc, + mapAliases, mapJoinPos, noCheckOuterJoin, true); + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java index bca91dd..b546838 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java @@ -21,12 +21,15 @@ package org.apache.hadoop.hive.ql.optimizer; import java.util.ArrayList; import java.util.EnumSet; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.Stack; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator; import org.apache.hadoop.hive.ql.exec.MapJoinOperator; import org.apache.hadoop.hive.ql.exec.Operator; @@ -58,11 +61,13 @@ import org.apache.hadoop.hive.ql.plan.TezWork; import org.apache.hadoop.hive.ql.plan.TezWork.VertexType; import org.apache.hadoop.hive.ql.stats.StatsUtils; +import com.google.common.collect.Sets; + import static org.apache.hadoop.hive.ql.plan.ReduceSinkDesc.ReducerTraits.FIXED; public class ReduceSinkMapJoinProc implements NodeProcessor { - protected transient Log LOG = LogFactory.getLog(this.getClass().getName()); + private final static Log LOG = LogFactory.getLog(ReduceSinkMapJoinProc.class.getName()); /* (non-Javadoc) * This processor addresses the RS-MJ case that occurs in tez on the small/hash @@ -79,7 +84,40 @@ public class ReduceSinkMapJoinProc implements NodeProcessor { GenTezProcContext context = (GenTezProcContext) procContext; MapJoinOperator mapJoinOp = (MapJoinOperator)nd; - if (stack.size() < 2 || !(stack.get(stack.size() - 2) instanceof ReduceSinkOperator)) { + // remember the original parent list before we start modifying it. + if (!context.mapJoinParentMap.containsKey(mapJoinOp)) { + List<Operator<?>> parents = new ArrayList<Operator<?>>(mapJoinOp.getParentOperators()); + context.mapJoinParentMap.put(mapJoinOp, parents); + } + + boolean isBigTable = stack.size() < 2 + || !(stack.get(stack.size() - 2) instanceof ReduceSinkOperator); + + ReduceSinkOperator parentRS = null; + if (!isBigTable) { + parentRS = (ReduceSinkOperator)stack.get(stack.size() - 2); + + // For dynamic partitioned hash join, the big table will also be coming from a ReduceSinkOperator + // Check for this condition. + // TODO: use indexOf(), or parentRS.getTag()? + isBigTable = + (mapJoinOp.getParentOperators().indexOf(parentRS) == mapJoinOp.getConf().getPosBigTable()); + } + + if (mapJoinOp.getConf().isDynamicPartitionHashJoin() && + !context.mapJoinToUnprocessedSmallTableReduceSinks.containsKey(mapJoinOp)) { + // Initialize set of unprocessed small tables + Set<ReduceSinkOperator> rsSet = Sets.newIdentityHashSet(); + for (int pos = 0; pos < mapJoinOp.getParentOperators().size(); ++pos) { + if (pos == mapJoinOp.getConf().getPosBigTable()) { + continue; + } + rsSet.add((ReduceSinkOperator) mapJoinOp.getParentOperators().get(pos)); + } + context.mapJoinToUnprocessedSmallTableReduceSinks.put(mapJoinOp, rsSet); + } + + if (isBigTable) { context.currentMapJoinOperators.add(mapJoinOp); return null; } @@ -87,14 +125,29 @@ public class ReduceSinkMapJoinProc implements NodeProcessor { context.preceedingWork = null; context.currentRootOperator = null; - ReduceSinkOperator parentRS = (ReduceSinkOperator)stack.get(stack.size() - 2); + return processReduceSinkToHashJoin(parentRS, mapJoinOp, context); + } + + public static BaseWork getMapJoinParentWork(GenTezProcContext context, Operator<?> parentRS) { + BaseWork parentWork; + if (context.unionWorkMap.containsKey(parentRS)) { + parentWork = context.unionWorkMap.get(parentRS); + } else { + assert context.childToWorkMap.get(parentRS).size() == 1; + parentWork = context.childToWorkMap.get(parentRS).get(0); + } + return parentWork; + } + + public static Object processReduceSinkToHashJoin(ReduceSinkOperator parentRS, MapJoinOperator mapJoinOp, + GenTezProcContext context) throws SemanticException { // remove the tag for in-memory side of mapjoin parentRS.getConf().setSkipTag(true); parentRS.setSkipTag(true); - // remember the original parent list before we start modifying it. - if (!context.mapJoinParentMap.containsKey(mapJoinOp)) { - List<Operator<?>> parents = new ArrayList<Operator<?>>(mapJoinOp.getParentOperators()); - context.mapJoinParentMap.put(mapJoinOp, parents); + + // Mark this small table as being processed + if (mapJoinOp.getConf().isDynamicPartitionHashJoin()) { + context.mapJoinToUnprocessedSmallTableReduceSinks.get(mapJoinOp).remove(parentRS); } List<BaseWork> mapJoinWork = null; @@ -109,13 +162,7 @@ public class ReduceSinkMapJoinProc implements NodeProcessor { * */ mapJoinWork = context.mapJoinWorkMap.get(mapJoinOp); - BaseWork parentWork; - if (context.unionWorkMap.containsKey(parentRS)) { - parentWork = context.unionWorkMap.get(parentRS); - } else { - assert context.childToWorkMap.get(parentRS).size() == 1; - parentWork = context.childToWorkMap.get(parentRS).get(0); - } + BaseWork parentWork = getMapJoinParentWork(context, parentRS); // set the link between mapjoin and parent vertex int pos = context.mapJoinParentMap.get(mapJoinOp).indexOf(parentRS); @@ -161,6 +208,11 @@ public class ReduceSinkMapJoinProc implements NodeProcessor { keyCount /= bucketCount; tableSize /= bucketCount; } + } else if (joinConf.isDynamicPartitionHashJoin()) { + // For dynamic partitioned hash join, assuming table is split evenly among the reduce tasks. + bucketCount = parentRS.getConf().getNumReducers(); + keyCount /= bucketCount; + tableSize /= bucketCount; } } LOG.info("Mapjoin " + mapJoinOp + ", pos: " + pos + " --> " + parentWork.getName() + " (" @@ -218,6 +270,8 @@ public class ReduceSinkMapJoinProc implements NodeProcessor { edgeType = EdgeType.CUSTOM_SIMPLE_EDGE; } } + } else if (mapJoinOp.getConf().isDynamicPartitionHashJoin()) { + edgeType = EdgeType.CUSTOM_SIMPLE_EDGE; } TezEdgeProperty edgeProp = new TezEdgeProperty(null, edgeType, numBuckets); @@ -232,7 +286,7 @@ public class ReduceSinkMapJoinProc implements NodeProcessor { } ReduceSinkOperator r = null; - if (parentRS.getConf().getOutputName() != null) { + if (context.connectedReduceSinks.contains(parentRS)) { LOG.debug("Cloning reduce sink for multi-child broadcast edge"); // we've already set this one up. Need to clone for the next work. r = (ReduceSinkOperator) OperatorFactory.getAndMakeChild( http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezProcContext.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezProcContext.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezProcContext.java index f474eae..9334c73 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezProcContext.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezProcContext.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hive.ql.parse; import java.io.Serializable; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.LinkedList; @@ -107,6 +108,10 @@ public class GenTezProcContext implements NodeProcessorCtx{ // map that says which mapjoin belongs to which work item public final Map<MapJoinOperator, List<BaseWork>> mapJoinWorkMap; + // Mapping of reducesink to mapjoin operators + // Only used for dynamic partitioned hash joins (mapjoin operator in the reducer) + public final Map<Operator<?>, MapJoinOperator> smallTableParentToMapJoinMap; + // a map to keep track of which root generated which work public final Map<Operator<?>, BaseWork> rootToWorkMap; @@ -151,6 +156,11 @@ public class GenTezProcContext implements NodeProcessorCtx{ // remember the connections between ts and event public final Map<TableScanOperator, List<AppMasterEventOperator>> tsToEventMap; + // When processing dynamic partitioned hash joins, some of the small tables may not get processed + // before the mapjoin's parents are removed during GenTezWork.process(). This is to keep + // track of which small tables haven't been processed yet. + public Map<MapJoinOperator, Set<ReduceSinkOperator>> mapJoinToUnprocessedSmallTableReduceSinks; + @SuppressWarnings("unchecked") public GenTezProcContext(HiveConf conf, ParseContext parseContext, List<Task<MoveWork>> moveTask, List<Task<? extends Serializable>> rootTasks, @@ -167,6 +177,7 @@ public class GenTezProcContext implements NodeProcessorCtx{ this.leafOperatorToFollowingWork = new LinkedHashMap<Operator<?>, BaseWork>(); this.linkOpWithWorkMap = new LinkedHashMap<Operator<?>, Map<BaseWork, TezEdgeProperty>>(); this.linkWorkWithReduceSinkMap = new LinkedHashMap<BaseWork, List<ReduceSinkOperator>>(); + this.smallTableParentToMapJoinMap = new LinkedHashMap<Operator<?>, MapJoinOperator>(); this.mapJoinWorkMap = new LinkedHashMap<MapJoinOperator, List<BaseWork>>(); this.rootToWorkMap = new LinkedHashMap<Operator<?>, BaseWork>(); this.childToWorkMap = new LinkedHashMap<Operator<?>, List<BaseWork>>(); @@ -188,6 +199,7 @@ public class GenTezProcContext implements NodeProcessorCtx{ this.tsToEventMap = new LinkedHashMap<TableScanOperator, List<AppMasterEventOperator>>(); this.opMergeJoinWorkMap = new LinkedHashMap<Operator<?>, MergeJoinWork>(); this.currentMergeJoinOperator = null; + this.mapJoinToUnprocessedSmallTableReduceSinks = new HashMap<MapJoinOperator, Set<ReduceSinkOperator>>(); rootTasks.add(currentTask); } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java index 93ad145..a9d1f8e 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezUtils.java @@ -35,6 +35,7 @@ import org.apache.hadoop.hive.ql.exec.AppMasterEventOperator; import org.apache.hadoop.hive.ql.exec.FetchTask; import org.apache.hadoop.hive.ql.exec.FileSinkOperator; import org.apache.hadoop.hive.ql.exec.HashTableDummyOperator; +import org.apache.hadoop.hive.ql.exec.MapJoinOperator; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.OperatorUtils; import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator; @@ -130,12 +131,13 @@ public class GenTezUtils { tezWork.add(reduceWork); TezEdgeProperty edgeProp; + EdgeType edgeType = determineEdgeType(context.preceedingWork, reduceWork); if (reduceWork.isAutoReduceParallelism()) { edgeProp = - new TezEdgeProperty(context.conf, EdgeType.SIMPLE_EDGE, true, + new TezEdgeProperty(context.conf, edgeType, true, reduceWork.getMinReduceTasks(), reduceWork.getMaxReduceTasks(), bytesPerReducer); } else { - edgeProp = new TezEdgeProperty(EdgeType.SIMPLE_EDGE); + edgeProp = new TezEdgeProperty(edgeType); } tezWork.connect( @@ -470,4 +472,21 @@ public class GenTezUtils { curr.removeChild(child); } + + public static EdgeType determineEdgeType(BaseWork preceedingWork, BaseWork followingWork) { + if (followingWork instanceof ReduceWork) { + // Ideally there should be a better way to determine that the followingWork contains + // a dynamic partitioned hash join, but in some cases (createReduceWork()) it looks like + // the work must be created/connected first, before the GenTezProcContext can be updated + // with the mapjoin/work relationship. + ReduceWork reduceWork = (ReduceWork) followingWork; + if (reduceWork.getReducer() instanceof MapJoinOperator) { + MapJoinOperator joinOp = (MapJoinOperator) reduceWork.getReducer(); + if (joinOp.getConf().isDynamicPartitionHashJoin()) { + return EdgeType.CUSTOM_SIMPLE_EDGE; + } + } + } + return EdgeType.SIMPLE_EDGE; + } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezWork.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezWork.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezWork.java index 6b3e19d..c4e0413 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezWork.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/GenTezWork.java @@ -40,6 +40,7 @@ import org.apache.hadoop.hive.ql.lib.Node; import org.apache.hadoop.hive.ql.lib.NodeProcessor; import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx; import org.apache.hadoop.hive.ql.optimizer.GenMapRedUtils; +import org.apache.hadoop.hive.ql.optimizer.ReduceSinkMapJoinProc; import org.apache.hadoop.hive.ql.plan.BaseWork; import org.apache.hadoop.hive.ql.plan.MergeJoinWork; import org.apache.hadoop.hive.ql.plan.OperatorDesc; @@ -165,8 +166,11 @@ public class GenTezWork implements NodeProcessor { mergeJoinWork.addMergedWork(work, null, context.leafOperatorToFollowingWork); Operator<? extends OperatorDesc> parentOp = getParentFromStack(context.currentMergeJoinOperator, stack); + // Set the big table position. Both the reduce work and merge join operator + // should be set with the same value. int pos = context.currentMergeJoinOperator.getTagForOperator(parentOp); work.setTag(pos); + context.currentMergeJoinOperator.getConf().setBigTablePosition(pos); tezWork.setVertexType(work, VertexType.MULTI_INPUT_UNINITIALIZED_EDGES); for (BaseWork parentWork : tezWork.getParents(work)) { TezEdgeProperty edgeProp = tezWork.getEdgeProperty(parentWork, work); @@ -190,6 +194,50 @@ public class GenTezWork implements NodeProcessor { // remember which mapjoin operator links with which work if (!context.currentMapJoinOperators.isEmpty()) { for (MapJoinOperator mj: context.currentMapJoinOperators) { + // For dynamic partitioned hash join, ReduceSinkMapJoinProc rule may not get run for all + // of the ReduceSink parents, because the parents of the MapJoin operator get + // removed later on in this method. Keep track of the parent to mapjoin mapping + // so we can later run the same logic that is run in ReduceSinkMapJoinProc. + if (mj.getConf().isDynamicPartitionHashJoin()) { + // Since this is a dynamic partitioned hash join, the work for this join should be a ReduceWork + ReduceWork reduceWork = (ReduceWork) work; + int bigTablePosition = mj.getConf().getPosBigTable(); + reduceWork.setTag(bigTablePosition); + + // Use context.mapJoinParentMap to get the original RS parents, because + // the MapJoin's parents may have been replaced by dummy operator. + List<Operator<?>> mapJoinOriginalParents = context.mapJoinParentMap.get(mj); + if (mapJoinOriginalParents == null) { + throw new SemanticException("Unexpected error - context.mapJoinParentMap did not have an entry for " + mj); + } + for (int pos = 0; pos < mapJoinOriginalParents.size(); ++pos) { + // This processing only needs to happen for the small tables + if (pos == bigTablePosition) { + continue; + } + Operator<?> parentOp = mapJoinOriginalParents.get(pos); + context.smallTableParentToMapJoinMap.put(parentOp, mj); + + ReduceSinkOperator parentRS = (ReduceSinkOperator) parentOp; + + // TableDesc needed for dynamic partitioned hash join + GenMapRedUtils.setKeyAndValueDesc(reduceWork, parentRS); + + // For small table RS parents that have already been processed, we need to + // add the tag to the RS work to the reduce work that contains this map join. + // This was not being done for normal mapjoins, where the small table typically + // has its ReduceSink parent removed. + if (!context.mapJoinToUnprocessedSmallTableReduceSinks.get(mj).contains(parentRS)) { + // This reduce sink has been processed already, so the work for the parentRS exists + BaseWork parentWork = ReduceSinkMapJoinProc.getMapJoinParentWork(context, parentRS); + int tag = parentRS.getConf().getTag(); + tag = (tag == -1 ? 0 : tag); + reduceWork.getTagToInput().put(tag, parentWork.getName()); + } + + } + } + LOG.debug("Processing map join: " + mj); // remember the mapping in case we scan another branch of the // mapjoin later @@ -369,15 +417,44 @@ public class GenTezWork implements NodeProcessor { // remember the output name of the reduce sink rs.getConf().setOutputName(rWork.getName()); + // For dynamic partitioned hash join, run the ReduceSinkMapJoinProc logic for any + // ReduceSink parents that we missed. + MapJoinOperator mj = context.smallTableParentToMapJoinMap.get(rs); + if (mj != null) { + // Only need to run the logic for tables we missed + if (context.mapJoinToUnprocessedSmallTableReduceSinks.get(mj).contains(rs)) { + // ReduceSinkMapJoinProc logic does not work unless the ReduceSink is connected as + // a parent of the MapJoin, but at this point we have already removed all of the + // parents from the MapJoin. + // Try temporarily adding the RS as a parent + ArrayList<Operator<?>> tempMJParents = new ArrayList<Operator<?>>(); + tempMJParents.add(rs); + mj.setParentOperators(tempMJParents); + // ReduceSink also needs MapJoin as child + List<Operator<?>> rsChildren = rs.getChildOperators(); + rsChildren.add(mj); + + // Since the MapJoin has had all of its other parents removed at this point, + // it would be bad here if processReduceSinkToHashJoin() tries to do anything + // with the RS parent based on its position in the list of parents. + ReduceSinkMapJoinProc.processReduceSinkToHashJoin(rs, mj, context); + + // Remove any parents from MapJoin again + mj.removeParents(); + // TODO: do we also need to remove the MapJoin from the list of RS's children? + } + } + if (!context.connectedReduceSinks.contains(rs)) { // add dependency between the two work items TezEdgeProperty edgeProp; + EdgeType edgeType = utils.determineEdgeType(work, followingWork); if (rWork.isAutoReduceParallelism()) { edgeProp = - new TezEdgeProperty(context.conf, EdgeType.SIMPLE_EDGE, true, + new TezEdgeProperty(context.conf, edgeType, true, rWork.getMinReduceTasks(), rWork.getMaxReduceTasks(), bytesPerReducer); } else { - edgeProp = new TezEdgeProperty(EdgeType.SIMPLE_EDGE); + edgeProp = new TezEdgeProperty(edgeType); } tezWork.connect(work, followingWork, edgeProp); context.connectedReduceSinks.add(rs); http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java index fa697ef..d574c5c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/BaseWork.java @@ -47,7 +47,7 @@ public abstract class BaseWork extends AbstractOperatorDesc { // Their function is mainly as root ops to give the mapjoin the correct // schema info. List<HashTableDummyOperator> dummyOps; - int tag; + int tag = 0; private final List<String> sortColNames = new ArrayList<String>(); private MapredLocalWork mrLocalWork; http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/plan/CommonMergeJoinDesc.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/CommonMergeJoinDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/CommonMergeJoinDesc.java index f9c34cb..cce9bc4 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/CommonMergeJoinDesc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/CommonMergeJoinDesc.java @@ -45,4 +45,8 @@ public class CommonMergeJoinDesc extends MapJoinDesc implements Serializable { public int getBigTablePosition() { return mapJoinConversionPos; } + + public void setBigTablePosition(int pos) { + mapJoinConversionPos = pos; + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java index fb3c4a3..e291a48 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/ExprNodeDescUtils.java @@ -28,6 +28,7 @@ import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluatorFactory; import org.apache.hadoop.hive.ql.exec.FunctionRegistry; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.UDF; +import org.apache.hadoop.hive.ql.optimizer.ConstantPropagateProcFactory; import org.apache.hadoop.hive.ql.parse.SemanticException; import org.apache.hadoop.hive.ql.udf.generic.GenericUDF; import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge; @@ -278,6 +279,59 @@ public class ExprNodeDescUtils { throw new SemanticException("Met multiple parent operators"); } + public static List<ExprNodeDesc> resolveJoinKeysAsRSColumns(List<ExprNodeDesc> sourceList, + Operator<?> reduceSinkOp) { + ArrayList<ExprNodeDesc> result = new ArrayList<ExprNodeDesc>(sourceList.size()); + for (ExprNodeDesc source : sourceList) { + ExprNodeDesc newExpr = resolveJoinKeysAsRSColumns(source, reduceSinkOp); + if (newExpr == null) { + return null; + } + result.add(newExpr); + } + return result; + } + + /** + * Join keys are expressions based on the select operator. Resolve the expressions so they + * are based on the ReduceSink operator + * SEL -> RS -> JOIN + * @param source + * @param reduceSinkOp + * @return + */ + public static ExprNodeDesc resolveJoinKeysAsRSColumns(ExprNodeDesc source, Operator<?> reduceSinkOp) { + // Assuming this is only being done for join keys. As a result we shouldn't have to recursively + // check any nested child expressions, because the result of the expression should exist as an + // output column of the ReduceSink operator + if (source == null) { + return null; + } + + // columnExprMap has the reverse of what we need - a mapping of the internal column names + // to the ExprNodeDesc from the previous operation. + // Find the key/value where the ExprNodeDesc value matches the column we are searching for. + // The key portion of the entry will be the internal column name for the join key expression. + for (Map.Entry<String, ExprNodeDesc> mapEntry : reduceSinkOp.getColumnExprMap().entrySet()) { + if (mapEntry.getValue().isSame(source)) { + String columnInternalName = mapEntry.getKey(); + if (source instanceof ExprNodeColumnDesc) { + // The join key is a table column. Create the ExprNodeDesc based on this column. + ColumnInfo columnInfo = reduceSinkOp.getSchema().getColumnInfo(columnInternalName); + return new ExprNodeColumnDesc(columnInfo); + } else { + // Join key expression is likely some expression involving functions/operators, so there + // is no actual table column for this. But the ReduceSink operator should still have an + // output column corresponding to this expression, using the columnInternalName. + // TODO: does tableAlias matter for this kind of expression? + return new ExprNodeColumnDesc(source.getTypeInfo(), columnInternalName, "", false); + } + } + } + + return null; // Couldn't find reference to expression + } + public static ExprNodeDesc[] extractComparePair(ExprNodeDesc expr1, ExprNodeDesc expr2) { expr1 = extractConstant(expr1); expr2 = extractConstant(expr2); @@ -483,4 +537,65 @@ public class ExprNodeDescUtils { return exprColLst; } + + public static List<ExprNodeDesc> flattenExprList(List<ExprNodeDesc> sourceList) { + ArrayList<ExprNodeDesc> result = new ArrayList<ExprNodeDesc>(sourceList.size()); + for (ExprNodeDesc source : sourceList) { + result.add(flattenExpr(source)); + } + return result; + } + + /** + * A normal reduce operator's rowObjectInspector looks like a struct containing + * nested key/value structs that contain the column values: + * { key: { reducesinkkey0:int }, value: { _col0:int, _col1:int, .. } } + * + * While the rowObjectInspector looks the same for vectorized queries during + * compilation time, within the tasks at query execution the rowObjectInspector + * has changed to a flatter structure without nested key/value structs: + * { 'key.reducesinkkey0':int, 'value._col0':int, 'value._col1':int, .. } + * + * Trying to fetch 'key.reducesinkkey0' by name from the list of flattened + * ObjectInspectors does not work because the '.' gets interpreted as a field member, + * even though it is a flattened list of column values. + * This workaround converts the column name referenced in the ExprNodeDesc + * from a nested field name (key.reducesinkkey0) to key_reducesinkkey0, + * simply by replacing '.' with '_'. + * @param source + * @return + */ + public static ExprNodeDesc flattenExpr(ExprNodeDesc source) { + if (source instanceof ExprNodeGenericFuncDesc) { + // all children expression should be resolved + ExprNodeGenericFuncDesc function = (ExprNodeGenericFuncDesc) source.clone(); + List<ExprNodeDesc> newChildren = flattenExprList(function.getChildren()); + for (ExprNodeDesc newChild : newChildren) { + if (newChild == null) { + // Could not resolve all of the function children, fail + return null; + } + } + function.setChildren(newChildren); + return function; + } + if (source instanceof ExprNodeColumnDesc) { + ExprNodeColumnDesc column = (ExprNodeColumnDesc) source; + // Create a new ColumnInfo, replacing STRUCT.COLUMN with STRUCT_COLUMN + String newColumn = column.getColumn().replace('.', '_'); + return new ExprNodeColumnDesc(source.getTypeInfo(), newColumn, column.getTabAlias(), false); + } + if (source instanceof ExprNodeFieldDesc) { + // field expression should be resolved + ExprNodeFieldDesc field = (ExprNodeFieldDesc) source.clone(); + ExprNodeDesc fieldDesc = flattenExpr(field.getDesc()); + if (fieldDesc == null) { + return null; + } + field.setDesc(fieldDesc); + return field; + } + // constant or null expr, just return + return source; + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/plan/MapJoinDesc.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/MapJoinDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/MapJoinDesc.java index cee9100..e27b89b 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/MapJoinDesc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/MapJoinDesc.java @@ -29,6 +29,8 @@ import java.util.Map.Entry; import java.util.Set; import org.apache.hadoop.hive.ql.plan.Explain.Level; +import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator; + /** * Map Join operator Descriptor implementation. * @@ -71,6 +73,7 @@ public class MapJoinDesc extends JoinDesc implements Serializable { protected boolean genJoinKeys = true; private boolean isHybridHashJoin; + private boolean isDynamicPartitionHashJoin = false; // Extra parameters only for vectorization. private VectorMapJoinDesc vectorDesc; @@ -369,4 +372,12 @@ public class MapJoinDesc extends JoinDesc implements Serializable { public boolean getGenJoinKeys() { return genJoinKeys; } + + public boolean isDynamicPartitionHashJoin() { + return isDynamicPartitionHashJoin; + } + + public void setDynamicPartitionHashJoin(boolean isDistributedHashJoin) { + this.isDynamicPartitionHashJoin = isDistributedHashJoin; + } } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceWork.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceWork.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceWork.java index a78a92e..020d6de 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceWork.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceWork.java @@ -136,7 +136,7 @@ public class ReduceWork extends BaseWork { return null; } if (valueObjectInspector == null) { - valueObjectInspector = getObjectInspector(tagToValueDesc.get(0)); + valueObjectInspector = getObjectInspector(tagToValueDesc.get(tag)); } return valueObjectInspector; } http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_1.q ---------------------------------------------------------------------- diff --git a/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_1.q b/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_1.q new file mode 100644 index 0000000..e3325c4 --- /dev/null +++ b/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_1.q @@ -0,0 +1,101 @@ + +set hive.explain.user=false; +set hive.auto.convert.join=false; +set hive.optimize.dynamic.partition.hashjoin=false; + +-- First try with regular mergejoin +explain +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +explain +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +explain +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; + +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; + +set hive.auto.convert.join=true; +set hive.optimize.dynamic.partition.hashjoin=true; +set hive.auto.convert.join.noconditionaltask.size=200000; +set hive.exec.reducers.bytes.per.reducer=200000; + +-- Try with dynamically partitioned hashjoin +explain +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +explain +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +explain +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; + +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_2.q ---------------------------------------------------------------------- diff --git a/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_2.q b/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_2.q new file mode 100644 index 0000000..af4e2b8 --- /dev/null +++ b/ql/src/test/queries/clientpositive/tez_dynpart_hashjoin_2.q @@ -0,0 +1,83 @@ + +set hive.explain.user=false; +set hive.auto.convert.join=false; +set hive.optimize.dynamic.partition.hashjoin=false; + +-- Multiple tables, and change the order of the big table (alltypesorc) +-- First try with regular mergejoin +explain +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +set hive.auto.convert.join=true; +set hive.optimize.dynamic.partition.hashjoin=true; +set hive.auto.convert.join.noconditionaltask.size=2000; +set hive.exec.reducers.bytes.per.reducer=200000; + +-- noconditionaltask.size needs to be low enough that entire filtered table results do not fit in one task's hash table +-- Try with dynamically partitioned hash join +explain +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +-- Try different order of tables +explain +select + a.* +from + src b, + alltypesorc a, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +select + a.* +from + src b, + alltypesorc a, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_1.q ---------------------------------------------------------------------- diff --git a/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_1.q b/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_1.q new file mode 100644 index 0000000..65fee16 --- /dev/null +++ b/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_1.q @@ -0,0 +1,102 @@ + +set hive.explain.user=false; +set hive.auto.convert.join=false; +set hive.optimize.dynamic.partition.hashjoin=false; + +-- First try with regular mergejoin +explain +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +explain +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +explain +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; + +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; + +set hive.auto.convert.join=true; +set hive.optimize.dynamic.partition.hashjoin=true; +set hive.auto.convert.join.noconditionaltask.size=200000; +set hive.exec.reducers.bytes.per.reducer=200000; +set hive.vectorized.execution.enabled=true; + +-- Try with dynamically partitioned hashjoin +explain +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +select + * +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +order by a.cint; + +explain +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +select + count(*) +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null; + +explain +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; + +select + a.csmallint, count(*) c1 +from alltypesorc a join alltypesorc b on a.cint = b.cint +where + a.cint between 1000000 and 3000000 and b.cbigint is not null +group by a.csmallint +order by c1; http://git-wip-us.apache.org/repos/asf/hive/blob/04d54f61/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_2.q ---------------------------------------------------------------------- diff --git a/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_2.q b/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_2.q new file mode 100644 index 0000000..606f455 --- /dev/null +++ b/ql/src/test/queries/clientpositive/tez_vector_dynpart_hashjoin_2.q @@ -0,0 +1,84 @@ + +set hive.explain.user=false; +set hive.auto.convert.join=false; +set hive.optimize.dynamic.partition.hashjoin=false; + +-- Multiple tables, and change the order of the big table (alltypesorc) +-- First try with regular mergejoin +explain +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +set hive.auto.convert.join=true; +set hive.optimize.dynamic.partition.hashjoin=true; +set hive.auto.convert.join.noconditionaltask.size=2000; +set hive.exec.reducers.bytes.per.reducer=200000; +set hive.vectorized.execution.enabled=true; + +-- noconditionaltask.size needs to be low enough that entire filtered table results do not fit in one task's hash table +-- Try with dynamically partitioned hash join +explain +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +select + a.* +from + alltypesorc a, + src b, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +-- Try different order of tables +explain +select + a.* +from + src b, + alltypesorc a, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint; + +select + a.* +from + src b, + alltypesorc a, + src c +where + a.csmallint = cast(b.key as int) and a.csmallint = (cast(c.key as int) + 0) + and (a.csmallint < 100) +order by a.csmallint, a.ctinyint, a.cint;
