[ https://issues.apache.org/jira/browse/HIVE-21196?focusedWorklogId=468101&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-468101 ]
ASF GitHub Bot logged work on HIVE-21196: ----------------------------------------- Author: ASF GitHub Bot Created on: 07/Aug/20 22:55 Start Date: 07/Aug/20 22:55 Worklog Time Spent: 10m Work Description: zabetak commented on a change in pull request #1325: URL: https://github.com/apache/hive/pull/1325#discussion_r467318976 ########## File path: ql/src/java/org/apache/hadoop/hive/ql/optimizer/SemiJoinReductionMerge.java ########## @@ -0,0 +1,399 @@ +/* + * 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.optimizer; + +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.ql.exec.ColumnInfo; +import org.apache.hadoop.hive.ql.exec.FilterOperator; +import org.apache.hadoop.hive.ql.exec.GroupByOperator; +import org.apache.hadoop.hive.ql.exec.Operator; +import org.apache.hadoop.hive.ql.exec.OperatorFactory; +import org.apache.hadoop.hive.ql.exec.OperatorUtils; +import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator; +import org.apache.hadoop.hive.ql.exec.RowSchema; +import org.apache.hadoop.hive.ql.exec.SelectOperator; +import org.apache.hadoop.hive.ql.exec.TableScanOperator; +import org.apache.hadoop.hive.ql.exec.Utilities; +import org.apache.hadoop.hive.ql.io.AcidUtils; +import org.apache.hadoop.hive.ql.parse.GenTezUtils; +import org.apache.hadoop.hive.ql.parse.ParseContext; +import org.apache.hadoop.hive.ql.parse.RuntimeValuesInfo; +import org.apache.hadoop.hive.ql.parse.SemanticAnalyzer; +import org.apache.hadoop.hive.ql.parse.SemanticException; +import org.apache.hadoop.hive.ql.parse.SemiJoinBranchInfo; +import org.apache.hadoop.hive.ql.plan.AggregationDesc; +import org.apache.hadoop.hive.ql.plan.DynamicValue; +import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeDynamicValueDesc; +import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc; +import org.apache.hadoop.hive.ql.plan.FilterDesc; +import org.apache.hadoop.hive.ql.plan.GroupByDesc; +import org.apache.hadoop.hive.ql.plan.PlanUtils; +import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc; +import org.apache.hadoop.hive.ql.plan.SelectDesc; +import org.apache.hadoop.hive.ql.plan.TableDesc; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFBloomFilter; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMax; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFMin; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBetween; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFInBloomFilter; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFMurmurHash; +import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPAnd; +import org.apache.hadoop.hive.ql.util.NullOrdering; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo; +import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.SortedMap; +import java.util.TreeMap; + +public class SemiJoinReductionMerge extends Transform { + + public ParseContext transform(ParseContext parseContext) throws SemanticException { + Map<ReduceSinkOperator, SemiJoinBranchInfo> map = parseContext.getRsToSemiJoinBranchInfo(); + if (map.isEmpty()) { + return parseContext; + } + HiveConf hiveConf = parseContext.getConf(); + + // Order does not really matter but it is necessary to keep plans stable + SortedMap<SJSourceTarget, List<ReduceSinkOperator>> sameTableSJ = + new TreeMap<>(Comparator.comparing(SJSourceTarget::toString)); + for (Map.Entry<ReduceSinkOperator, SemiJoinBranchInfo> smjEntry : map.entrySet()) { + TableScanOperator ts = smjEntry.getValue().getTsOp(); + // Semijoin optimization branch should look like <Parent>-SEL-GB1-RS1-GB2-RS2 + SelectOperator selOp = OperatorUtils.ancestor(smjEntry.getKey(), SelectOperator.class, 0, 0, 0, 0); + assert selOp != null; + assert selOp.getParentOperators().size() == 1; + Operator<?> source = selOp.getParentOperators().get(0); + SJSourceTarget sjKey = new SJSourceTarget(source, ts); + List<ReduceSinkOperator> ops = sameTableSJ.computeIfAbsent(sjKey, tableScanOperator -> new ArrayList<>()); + ops.add(smjEntry.getKey()); + } + for (Map.Entry<SJSourceTarget, List<ReduceSinkOperator>> sjMergeCandidate : sameTableSJ.entrySet()) { + final List<ReduceSinkOperator> sjBrances = sjMergeCandidate.getValue(); + if (sjBrances.size() < 2) { + continue; + } + // Order does not really matter but it is necessary to keep plans stable + sjBrances.sort(Comparator.comparing(Operator::getIdentifier)); + + List<SelectOperator> selOps = new ArrayList<>(sjBrances.size()); + for (ReduceSinkOperator rs : sjBrances) { + selOps.add(OperatorUtils.ancestor(rs, SelectOperator.class, 0, 0, 0, 0)); + } + SelectOperator selectOp = mergeSelectOps(sjMergeCandidate.getKey().source, selOps); + + GroupByOperator gbPartialOp = createGroupBy(selectOp, selectOp, GroupByDesc.Mode.HASH, hiveConf); + + ReduceSinkOperator rsPartialOp = createReduceSink(gbPartialOp, NullOrdering.defaultNullOrder(hiveConf)); + rsPartialOp.getConf().setReducerTraits(EnumSet.of(ReduceSinkDesc.ReducerTraits.QUICKSTART)); + + GroupByOperator gbCompleteOp = createGroupBy(selectOp, rsPartialOp, GroupByDesc.Mode.FINAL, hiveConf); + + ReduceSinkOperator rsCompleteOp = createReduceSink(gbCompleteOp, NullOrdering.defaultNullOrder(hiveConf)); + + final TableScanOperator sjTargetTable = sjMergeCandidate.getKey().target; + SemiJoinBranchInfo sjInfo = new SemiJoinBranchInfo(sjTargetTable, false); + parseContext.getRsToSemiJoinBranchInfo().put(rsCompleteOp, sjInfo); + + // Save the info that is required at query time to resolve dynamic/runtime values. + RuntimeValuesInfo valuesInfo = createRuntimeValuesInfo(rsCompleteOp, sjBrances, parseContext); + parseContext.getRsToRuntimeValuesInfoMap().put(rsCompleteOp, valuesInfo); + + ExprNodeGenericFuncDesc sjPredicate = createSemiJoinPredicate(sjBrances, valuesInfo, parseContext); + + // Update filter operators with the new semi-join predicate + for (Operator<?> op : sjTargetTable.getChildOperators()) { + if (op instanceof FilterOperator) { + FilterDesc filter = ((FilterOperator) op).getConf(); + filter.setPredicate(and(Arrays.asList(filter.getPredicate(), sjPredicate))); + } + } + // Update tableScan with the new semi-join predicate + sjTargetTable.getConf().setFilterExpr(and(Arrays.asList(sjTargetTable.getConf().getFilterExpr(), sjPredicate))); + + for (ReduceSinkOperator rs : sjBrances) { + GenTezUtils.removeSemiJoinOperator(parseContext, rs, sjTargetTable); + GenTezUtils.removeBranch(rs); + } + + // TODO How to associate multi-cols with gb ? + // parseContext.getColExprToGBMap().put(key, gb); + } + return parseContext; + } + + private static ExprNodeGenericFuncDesc createSemiJoinPredicate(List<ReduceSinkOperator> sjBranches, + RuntimeValuesInfo sjValueInfo, ParseContext context) { + // Performance note: To speed-up evaluation 'BETWEEN' predicates should come before the 'IN_BLOOM_FILTER' + Deque<String> dynamicIds = new ArrayDeque<>(sjValueInfo.getDynamicValueIDs()); + List<ExprNodeDesc> sjPredicates = new ArrayList<>(); + List<ExprNodeDesc> hashArgs = new ArrayList<>(); + for (ReduceSinkOperator rs : sjBranches) { + RuntimeValuesInfo info = context.getRsToRuntimeValuesInfoMap().get(rs); + assert info.getTargetColumns().size() == 1; + final ExprNodeDesc targetColumn = info.getTargetColumns().get(0); + TypeInfo typeInfo = targetColumn.getTypeInfo(); + DynamicValue minDynamic = new DynamicValue(dynamicIds.poll(), typeInfo); + DynamicValue maxDynamic = new DynamicValue(dynamicIds.poll(), typeInfo); + + List<ExprNodeDesc> betweenArgs = Arrays.asList( + // Use false to not invert between result + new ExprNodeConstantDesc(Boolean.FALSE), + targetColumn, + new ExprNodeDynamicValueDesc(minDynamic), + new ExprNodeDynamicValueDesc(maxDynamic)); + ExprNodeDesc betweenExp = + new ExprNodeGenericFuncDesc(TypeInfoFactory.booleanTypeInfo, new GenericUDFBetween(), "between", betweenArgs); + sjPredicates.add(betweenExp); + hashArgs.add(targetColumn); + } + + ExprNodeDesc hashExp = + new ExprNodeGenericFuncDesc(TypeInfoFactory.intTypeInfo, new GenericUDFMurmurHash(), "hash", hashArgs); + + assert dynamicIds.size() == 1 : "There should be one column left untreated the one with the bloom filter"; + DynamicValue bloomDynamic = new DynamicValue(dynamicIds.poll(), TypeInfoFactory.binaryTypeInfo); + sjPredicates.add( + new ExprNodeGenericFuncDesc(TypeInfoFactory.booleanTypeInfo, new GenericUDFInBloomFilter(), "in_bloom_filter", + Arrays.asList(hashExp, new ExprNodeDynamicValueDesc(bloomDynamic)))); + return and(sjPredicates); + } + + private static RuntimeValuesInfo createRuntimeValuesInfo(ReduceSinkOperator rs, List<ReduceSinkOperator> sjBranches, + ParseContext parseContext) { + List<ExprNodeDesc> valueCols = rs.getConf().getValueCols(); + RuntimeValuesInfo info = new RuntimeValuesInfo(); + TableDesc rsFinalTableDesc = + PlanUtils.getReduceValueTableDesc(PlanUtils.getFieldSchemasFromColumnList(valueCols, "_col")); + List<String> dynamicValueIDs = new ArrayList<>(); + for (ExprNodeDesc rsCol : valueCols) { + dynamicValueIDs.add(rs.toString() + rsCol.getExprString()); + } + + info.setTableDesc(rsFinalTableDesc); + info.setDynamicValueIDs(dynamicValueIDs); + info.setColExprs(valueCols); + List<ExprNodeDesc> targetTableExpressions = new ArrayList<>(); + for (ReduceSinkOperator sjBranch : sjBranches) { + RuntimeValuesInfo sjInfo = parseContext.getRsToRuntimeValuesInfoMap().get(sjBranch); + assert sjInfo.getTargetColumns().size() == 1; + targetTableExpressions.add(sjInfo.getTargetColumns().get(0)); + } + info.setTargetColumns(targetTableExpressions); + return info; + } + + private static SelectOperator mergeSelectOps(Operator<?> parent, List<SelectOperator> selectOperators) { + List<String> colNames = new ArrayList<>(); + List<ExprNodeDesc> colDescs = new ArrayList<>(); + List<ColumnInfo> columnInfos = new ArrayList<>(); + Map<String, ExprNodeDesc> selectColumnExprMap = new HashMap<>(); + for (SelectOperator sel : selectOperators) { + for (ExprNodeDesc col : sel.getConf().getColList()) { + String colName = HiveConf.getColumnInternalName(colDescs.size()); + colNames.add(colName); + columnInfos.add(new ColumnInfo(colName, col.getTypeInfo(), "", false)); + colDescs.add(col); + selectColumnExprMap.put(colName, col); + } + } + ExprNodeDesc hashExp = + new ExprNodeGenericFuncDesc(TypeInfoFactory.intTypeInfo, new GenericUDFMurmurHash(), "hash", colDescs); + String hashName = HiveConf.getColumnInternalName(colDescs.size() + 1); + colNames.add(hashName); + columnInfos.add(new ColumnInfo(hashName, hashExp.getTypeInfo(), "", false)); + + List<ExprNodeDesc> selDescs = new ArrayList<>(colDescs); + selDescs.add(hashExp); + + SelectDesc select = new SelectDesc(selDescs, colNames); + SelectOperator selectOp = + (SelectOperator) OperatorFactory.getAndMakeChild(select, new RowSchema(columnInfos), parent); + selectOp.setColumnExprMap(selectColumnExprMap); + return selectOp; + } + + private static ReduceSinkOperator createReduceSink(Operator<?> parentOp, NullOrdering nullOrder) + throws SemanticException { + List<ExprNodeDesc> valueCols = new ArrayList<>(); + RowSchema parentSchema = parentOp.getSchema(); + List<String> outColNames = new ArrayList<>(); + for (int i = 0; i < parentSchema.getSignature().size(); i++) { + ColumnInfo colInfo = parentSchema.getSignature().get(i); + ExprNodeColumnDesc colExpr = new ExprNodeColumnDesc(colInfo.getType(), colInfo.getInternalName(), "", false); + valueCols.add(colExpr); + outColNames.add(SemanticAnalyzer.getColumnInternalName(i)); + } + + ReduceSinkDesc rsDesc = PlanUtils + .getReduceSinkDesc(Collections.emptyList(), valueCols, outColNames, false, -1, 0, 1, + AcidUtils.Operation.NOT_ACID, nullOrder); + rsDesc.setColumnExprMap(Collections.emptyMap()); + return (ReduceSinkOperator) OperatorFactory.getAndMakeChild(rsDesc, new RowSchema(parentSchema), parentOp); + } + + private static GroupByOperator createGroupBy(SelectOperator selectOp, Operator<?> parentOp, GroupByDesc.Mode gbMode, + HiveConf hiveConf) { + + final List<ExprNodeDesc> params; + final GenericUDAFEvaluator.Mode udafMode = SemanticAnalyzer.groupByDescModeToUDAFMode(gbMode, false); + switch (gbMode) { + case FINAL: + params = createGroupByAggregationParameters((ReduceSinkOperator) parentOp); + break; + case HASH: + params = createGroupByAggregationParameters(selectOp); + break; + default: + throw new AssertionError(gbMode.toString() + " is not supported"); + } + + List<AggregationDesc> gbAggs = new ArrayList<>(); + Deque<ExprNodeDesc> paramsCopy = new ArrayDeque<>(params); + while (paramsCopy.size() > 1) { + gbAggs.add(minAggregation(udafMode, paramsCopy.poll())); + gbAggs.add(maxAggregation(udafMode, paramsCopy.poll())); + } + gbAggs.add(bloomFilterAggregation(udafMode, paramsCopy.poll(), selectOp, hiveConf)); + assert paramsCopy.size() == 0; + + List<String> gbOutputNames = new ArrayList<>(gbAggs.size()); + List<ColumnInfo> gbColInfos = new ArrayList<>(gbAggs.size()); + for (int i = 0; i < params.size(); i++) { + String colName = HiveConf.getColumnInternalName(i); + gbOutputNames.add(colName); + final TypeInfo colType; + if (i == params.size() - 1) { + colType = TypeInfoFactory.binaryTypeInfo; // Bloom type + } else { + colType = params.get(i).getTypeInfo(); // Min/Max type + } + gbColInfos.add(new ColumnInfo(colName, colType, "", false)); + } + + float groupByMemoryUsage = HiveConf.getFloatVar(hiveConf, HiveConf.ConfVars.HIVEMAPAGGRHASHMEMORY); + float memoryThreshold = HiveConf.getFloatVar(hiveConf, HiveConf.ConfVars.HIVEMAPAGGRMEMORYTHRESHOLD); + float minReductionHashAggr = HiveConf.getFloatVar(hiveConf, HiveConf.ConfVars.HIVEMAPAGGRHASHMINREDUCTION); + GroupByDesc groupBy = + new GroupByDesc(gbMode, gbOutputNames, Collections.emptyList(), gbAggs, false, groupByMemoryUsage, + memoryThreshold, minReductionHashAggr, null, false, -1, false); + groupBy.setColumnExprMap(Collections.emptyMap()); + return (GroupByOperator) OperatorFactory.getAndMakeChild(groupBy, new RowSchema(gbColInfos), parentOp); + } + + private static List<ExprNodeDesc> createGroupByAggregationParameters(SelectOperator selectOp) { + List<ExprNodeDesc> params = new ArrayList<>(); + // The first n-1 cols are used as parameters for min & max so we need two expressions + for (ColumnInfo c : selectOp.getSchema().getSignature()) { + String name = c.getInternalName(); + ExprNodeColumnDesc p = new ExprNodeColumnDesc(new ColumnInfo(name, c.getType(), "", false)); + params.add(p); + params.add(p); + } + // The last col is used as parameter for bloom so we need only one expression + params.remove(params.size() - 1); + return params; + } + + private static List<ExprNodeDesc> createGroupByAggregationParameters(ReduceSinkOperator reduceOp) { + List<ExprNodeDesc> params = new ArrayList<>(); + // There is a 1-1 mapping between columns and parameters for the aggregation functions min, max, bloom + for (ColumnInfo c : reduceOp.getSchema().getSignature()) { + String name = Utilities.ReduceField.VALUE + "." + c.getInternalName(); + params.add(new ExprNodeColumnDesc(new ColumnInfo(name, c.getType(), "", false))); + } + return params; + } + + private static ExprNodeGenericFuncDesc and(List<ExprNodeDesc> args) { + return new ExprNodeGenericFuncDesc(TypeInfoFactory.booleanTypeInfo, new GenericUDFOPAnd(), "and", args); + } + + private static AggregationDesc minAggregation(GenericUDAFEvaluator.Mode mode, ExprNodeDesc col) { + List<ExprNodeDesc> p = Collections.singletonList(col); + return new AggregationDesc("min", new GenericUDAFMin.GenericUDAFMinEvaluator(), p, false, mode); + } + + private static AggregationDesc maxAggregation(GenericUDAFEvaluator.Mode mode, ExprNodeDesc col) { + List<ExprNodeDesc> p = Collections.singletonList(col); + return new AggregationDesc("max", new GenericUDAFMax.GenericUDAFMaxEvaluator(), p, false, mode); + } + + private static AggregationDesc bloomFilterAggregation(GenericUDAFEvaluator.Mode mode, ExprNodeDesc col, + SelectOperator source, HiveConf conf) { + GenericUDAFBloomFilter.GenericUDAFBloomFilterEvaluator bloomFilterEval = + new GenericUDAFBloomFilter.GenericUDAFBloomFilterEvaluator(); + bloomFilterEval.setSourceOperator(source); + bloomFilterEval.setMaxEntries(conf.getLongVar(HiveConf.ConfVars.TEZ_MAX_BLOOM_FILTER_ENTRIES)); + bloomFilterEval.setMinEntries(conf.getLongVar(HiveConf.ConfVars.TEZ_MIN_BLOOM_FILTER_ENTRIES)); + bloomFilterEval.setFactor(conf.getFloatVar(HiveConf.ConfVars.TEZ_BLOOM_FILTER_FACTOR)); + // TODO Setup hints + List<ExprNodeDesc> p = Collections.singletonList(col); + AggregationDesc bloom = new AggregationDesc("bloom_filter", bloomFilterEval, p, false, mode); + // TODO Why do we need to set it explicitly? Review comment: Removed in https://github.com/apache/hive/pull/1325/commits/bf5cad3aa0861eef811565214231e5dec87cd3e6 and created https://issues.apache.org/jira/browse/HIVE-24018 if I ever feel like investigating further. ---------------------------------------------------------------- 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: us...@infra.apache.org Issue Time Tracking ------------------- Worklog Id: (was: 468101) Time Spent: 3h 20m (was: 3h 10m) > Support semijoin reduction on multiple column join > -------------------------------------------------- > > Key: HIVE-21196 > URL: https://issues.apache.org/jira/browse/HIVE-21196 > Project: Hive > Issue Type: Bug > Reporter: Deepak Jaiswal > Assignee: Stamatis Zampetakis > Priority: Major > Labels: pull-request-available > Time Spent: 3h 20m > Remaining Estimate: 0h > > Currently for a query involving join on multiple columns creates separate > semi join edges for each key which in turn create a bloom filter for each of > them, like below, > EXPLAIN select count(*) from srcpart_date_n7 join srcpart_small_n3 on > (srcpart_date_n7.key = srcpart_small_n3.key1 and srcpart_date_n7.value = > srcpart_small_n3.value1) > {code:java} > Map 1 <- Reducer 5 (BROADCAST_EDGE) > Reducer 2 <- Map 1 (SIMPLE_EDGE), Map 4 (SIMPLE_EDGE) > Reducer 3 <- Reducer 2 (CUSTOM_SIMPLE_EDGE) > Reducer 5 <- Map 4 (CUSTOM_SIMPLE_EDGE) > #### A masked pattern was here #### > Vertices: > Map 1 > Map Operator Tree: > TableScan > alias: srcpart_date_n7 > filterExpr: (key is not null and value is not null and (key > BETWEEN DynamicValue(RS_7_srcpart_small_n3_key1_min) AND > DynamicValue(RS_7_srcpart_small_n3_key1_max) and in_bloom_filter(key, > DynamicValue(RS_7_srcpart_small_n3_key1_bloom_filter)))) (type: boolean) > Statistics: Num rows: 2000 Data size: 356000 Basic stats: > COMPLETE Column stats: COMPLETE > Filter Operator > predicate: ((key BETWEEN > DynamicValue(RS_7_srcpart_small_n3_key1_min) AND > DynamicValue(RS_7_srcpart_small_n3_key1_max) and in_bloom_filter(key, > DynamicValue(RS_7_srcpart_small_n3_key1_bloom_filter))) and key is not null > and value is not null) (type: boolean) > Statistics: Num rows: 2000 Data size: 356000 Basic stats: > COMPLETE Column stats: COMPLETE > Select Operator > expressions: key (type: string), value (type: string) > outputColumnNames: _col0, _col1 > Statistics: Num rows: 2000 Data size: 356000 Basic > stats: COMPLETE Column stats: COMPLETE > Reduce Output Operator > key expressions: _col0 (type: string), _col1 (type: > string) > sort order: ++ > Map-reduce partition columns: _col0 (type: string), > _col1 (type: string) > Statistics: Num rows: 2000 Data size: 356000 Basic > stats: COMPLETE Column stats: COMPLETE > Execution mode: vectorized, llap > LLAP IO: all inputs > Map 4 > Map Operator Tree: > TableScan > alias: srcpart_small_n3 > filterExpr: (key1 is not null and value1 is not null) > (type: boolean) > Statistics: Num rows: 20 Data size: 3560 Basic stats: > PARTIAL Column stats: PARTIAL > Filter Operator > predicate: (key1 is not null and value1 is not null) > (type: boolean) > Statistics: Num rows: 20 Data size: 3560 Basic stats: > PARTIAL Column stats: PARTIAL > Select Operator > expressions: key1 (type: string), value1 (type: string) > outputColumnNames: _col0, _col1 > Statistics: Num rows: 20 Data size: 3560 Basic stats: > PARTIAL Column stats: PARTIAL > Reduce Output Operator > key expressions: _col0 (type: string), _col1 (type: > string) > sort order: ++ > Map-reduce partition columns: _col0 (type: string), > _col1 (type: string) > Statistics: Num rows: 20 Data size: 3560 Basic stats: > PARTIAL Column stats: PARTIAL > Select Operator > expressions: _col0 (type: string) > outputColumnNames: _col0 > Statistics: Num rows: 20 Data size: 3560 Basic stats: > PARTIAL Column stats: PARTIAL > Group By Operator > aggregations: min(_col0), max(_col0), > bloom_filter(_col0, expectedEntries=20) > mode: hash > outputColumnNames: _col0, _col1, _col2 > Statistics: Num rows: 1 Data size: 730 Basic stats: > PARTIAL Column stats: PARTIAL > Reduce Output Operator > sort order: > Statistics: Num rows: 1 Data size: 730 Basic > stats: PARTIAL Column stats: PARTIAL > value expressions: _col0 (type: string), _col1 > (type: string), _col2 (type: binary) > Execution mode: vectorized, llap > LLAP IO: all inputs > Reducer 2 > Execution mode: llap > Reduce Operator Tree: > Merge Join Operator > condition map: > Inner Join 0 to 1 > keys: > 0 _col0 (type: string), _col1 (type: string) > 1 _col0 (type: string), _col1 (type: string) > Statistics: Num rows: 2200 Data size: 391600 Basic stats: > PARTIAL Column stats: NONE > Group By Operator > aggregations: count() > mode: hash > outputColumnNames: _col0 > Statistics: Num rows: 1 Data size: 8 Basic stats: PARTIAL > Column stats: NONE > Reduce Output Operator > sort order: > Statistics: Num rows: 1 Data size: 8 Basic stats: PARTIAL > Column stats: NONE > value expressions: _col0 (type: bigint) > Reducer 3 > Execution mode: vectorized, llap > Reduce Operator Tree: > Group By Operator > aggregations: count(VALUE._col0) > mode: mergepartial > outputColumnNames: _col0 > Statistics: Num rows: 1 Data size: 8 Basic stats: PARTIAL > Column stats: NONE > File Output Operator > compressed: false > Statistics: Num rows: 1 Data size: 8 Basic stats: PARTIAL > Column stats: NONE > table: > input format: > org.apache.hadoop.mapred.SequenceFileInputFormat > output format: > org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat > serde: > org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe > Reducer 5 > Execution mode: vectorized, llap > Reduce Operator Tree: > Group By Operator > aggregations: min(VALUE._col0), max(VALUE._col1), > bloom_filter(VALUE._col2, expectedEntries=20) > mode: final > outputColumnNames: _col0, _col1, _col2 > Statistics: Num rows: 1 Data size: 730 Basic stats: PARTIAL > Column stats: PARTIAL > Reduce Output Operator > sort order: > Statistics: Num rows: 1 Data size: 730 Basic stats: PARTIAL > Column stats: PARTIAL > value expressions: _col0 (type: string), _col1 (type: > string), _col2 (type: binary) > {code} > Instead it should create one branch for a join with one bloom filter. > > The implementation for bloom filter requires getting a hash out of all the > key columns and converting it to a long and feeding it to bloom filter as > input. This requires a new UDF which does this. It will be called at both > bloom filter generation and lookup phases. > The min and max will stay independent as they are today for each columns. > A vectorized implementation of such UDF is also required. -- This message was sent by Atlassian Jira (v8.3.4#803005)