You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ha...@apache.org on 2014/07/19 19:17:43 UTC
svn commit: r1611924 - in /hive/trunk:
common/src/java/org/apache/hadoop/hive/conf/ itests/qtest/
ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/
ql/src/java/org/apache/hadoop/hive/ql/parse/
ql/src/test/queries/clientpositive/ ql/src/test/res...
Author: hashutosh
Date: Sat Jul 19 17:17:42 2014
New Revision: 1611924
URL: http://svn.apache.org/r1611924
Log:
HIVE-7385 : Optimize for empty relation scans (Ashutosh Chauhan via Gunther Hagleitner)
Added:
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanOptimizer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java
hive/trunk/ql/src/test/queries/clientpositive/optimize_nullscan.q
hive/trunk/ql/src/test/results/clientpositive/optimize_nullscan.q.out
hive/trunk/ql/src/test/results/clientpositive/tez/optimize_nullscan.q.out
Modified:
hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
hive/trunk/itests/qtest/testconfiguration.properties
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MetadataOnlyOptimizer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java
Modified: hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
URL: http://svn.apache.org/viewvc/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java?rev=1611924&r1=1611923&r2=1611924&view=diff
==============================================================================
--- hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java (original)
+++ hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java Sat Jul 19 17:17:42 2014
@@ -984,6 +984,7 @@ public class HiveConf extends Configurat
HIVEPPDREMOVEDUPLICATEFILTERS("hive.ppd.remove.duplicatefilters", true,
"Whether to push predicates down into storage handlers. Ignored when hive.optimize.ppd is false."),
HIVEMETADATAONLYQUERIES("hive.optimize.metadataonly", true, ""),
+ HIVENULLSCANOPTIMIZE("hive.optimize.null.scan", true, "Dont scan relations which are guaranteed to not generate any rows"),
HIVEOPTPPD_STORAGE("hive.optimize.ppd.storage", true,
"Whether to push predicates down to storage handlers"),
HIVEOPTGROUPBY("hive.optimize.groupby", true,
@@ -1820,19 +1821,25 @@ public class HiveConf extends Configurat
enum VarType {
STRING {
+ @Override
void checkType(String value) throws Exception { }
+ @Override
String defaultValueString(ConfVars confVar) { return confVar.defaultStrVal; }
},
INT {
+ @Override
void checkType(String value) throws Exception { Integer.valueOf(value); }
},
LONG {
+ @Override
void checkType(String value) throws Exception { Long.valueOf(value); }
},
FLOAT {
+ @Override
void checkType(String value) throws Exception { Float.valueOf(value); }
},
BOOLEAN {
+ @Override
void checkType(String value) throws Exception { Boolean.valueOf(value); }
};
Modified: hive/trunk/itests/qtest/testconfiguration.properties
URL: http://svn.apache.org/viewvc/hive/trunk/itests/qtest/testconfiguration.properties?rev=1611924&r1=1611923&r2=1611924&view=diff
==============================================================================
--- hive/trunk/itests/qtest/testconfiguration.properties (original)
+++ hive/trunk/itests/qtest/testconfiguration.properties Sat Jul 19 17:17:42 2014
@@ -1,5 +1,5 @@
minimr.query.files=stats_counter_partitioned.q,list_bucket_dml_10.q,input16_cc.q,scriptfile1.q,scriptfile1_win.q,bucket4.q,bucketmapjoin6.q,disable_merge_for_bucketing.q,reduce_deduplicate.q,smb_mapjoin_8.q,join1.q,groupby2.q,bucketizedhiveinputformat.q,bucketmapjoin7.q,optrstat_groupby.q,bucket_num_reducers.q,bucket5.q,load_fs2.q,bucket_num_reducers2.q,infer_bucket_sort_merge.q,infer_bucket_sort_reducers_power_two.q,infer_bucket_sort_dyn_part.q,infer_bucket_sort_bucketed_table.q,infer_bucket_sort_map_operators.q,infer_bucket_sort_num_buckets.q,leftsemijoin_mr.q,schemeAuthority.q,schemeAuthority2.q,truncate_column_buckets.q,remote_script.q,,load_hdfs_file_with_space_in_the_name.q,parallel_orderby.q,import_exported_table.q,stats_counter.q,auto_sortmerge_join_16.q,quotedid_smb.q,file_with_header_footer.q,external_table_with_space_in_location_path.q,root_dir_external_table.q,index_bitmap3.q,ql_rewrite_gbtoidx.q,index_bitmap_auto.q,udf_using.q,empty_dir_in_table.q,temp_table_external.q
minimr.query.negative.files=cluster_tasklog_retrieval.q,minimr_broken_pipe.q,mapreduce_stack_trace.q,mapreduce_stack_trace_turnoff.q,mapreduce_stack_trace_hadoop20.q,mapreduce_stack_trace_turnoff_hadoop20.q,file_with_header_footer_negative.q,udf_local_resource.q
minitez.query.files=tez_fsstat.q,mapjoin_decimal.q,tez_join_tests.q,tez_joins_explain.q,mrr.q,tez_dml.q,tez_insert_overwrite_local_directory_1.q,tez_union.q,bucket_map_join_tez1.q,bucket_map_join_tez2.q,tez_schema_evolution.q,tez_join_hash.q
-minitez.query.files.shared=cross_product_check_1.q,cross_product_check_2.q,dynpart_sort_opt_vectorization.q,dynpart_sort_optimization.q,orc_analyze.q,join0.q,join1.q,auto_join0.q,auto_join1.q,bucket2.q,bucket3.q,bucket4.q,count.q,create_merge_compressed.q,cross_join.q,ctas.q,custom_input_output_format.q,disable_merge_for_bucketing.q,enforce_order.q,filter_join_breaktask.q,filter_join_breaktask2.q,groupby1.q,groupby2.q,groupby3.q,having.q,insert1.q,insert_into1.q,insert_into2.q,leftsemijoin.q,limit_pushdown.q,load_dyn_part1.q,load_dyn_part2.q,load_dyn_part3.q,mapjoin_mapjoin.q,mapreduce1.q,mapreduce2.q,merge1.q,merge2.q,metadata_only_queries.q,sample1.q,subquery_in.q,subquery_exists.q,vectorization_15.q,ptf.q,stats_counter.q,stats_noscan_1.q,stats_counter_partitioned.q,union2.q,union3.q,union4.q,union5.q,union6.q,union7.q,union8.q,union9.q,transform1.q,transform2.q,transform_ppr1.q,transform_ppr2.q,script_env_var1.q,script_env_var2.q,script_pipe.q,scriptfile1.q,metadataonly1.q,temp_t
able.q,vectorized_ptf.q
+minitez.query.files.shared=cross_product_check_1.q,cross_product_check_2.q,dynpart_sort_opt_vectorization.q,dynpart_sort_optimization.q,orc_analyze.q,join0.q,join1.q,auto_join0.q,auto_join1.q,bucket2.q,bucket3.q,bucket4.q,count.q,create_merge_compressed.q,cross_join.q,ctas.q,custom_input_output_format.q,disable_merge_for_bucketing.q,enforce_order.q,filter_join_breaktask.q,filter_join_breaktask2.q,groupby1.q,groupby2.q,groupby3.q,having.q,insert1.q,insert_into1.q,insert_into2.q,leftsemijoin.q,limit_pushdown.q,load_dyn_part1.q,load_dyn_part2.q,load_dyn_part3.q,mapjoin_mapjoin.q,mapreduce1.q,mapreduce2.q,merge1.q,merge2.q,metadata_only_queries.q,sample1.q,subquery_in.q,subquery_exists.q,vectorization_15.q,ptf.q,stats_counter.q,stats_noscan_1.q,stats_counter_partitioned.q,union2.q,union3.q,union4.q,union5.q,union6.q,union7.q,union8.q,union9.q,transform1.q,transform2.q,transform_ppr1.q,transform_ppr2.q,script_env_var1.q,script_env_var2.q,script_pipe.q,scriptfile1.q,metadataonly1.q,temp_t
able.q,vectorized_ptf.q,optimize_nullscan.q
beeline.positive.exclude=add_part_exist.q,alter1.q,alter2.q,alter4.q,alter5.q,alter_rename_partition.q,alter_rename_partition_authorization.q,archive.q,archive_corrupt.q,archive_multi.q,archive_mr_1806.q,archive_multi_mr_1806.q,authorization_1.q,authorization_2.q,authorization_4.q,authorization_5.q,authorization_6.q,authorization_7.q,ba_table1.q,ba_table2.q,ba_table3.q,ba_table_udfs.q,binary_table_bincolserde.q,binary_table_colserde.q,cluster.q,columnarserde_create_shortcut.q,combine2.q,constant_prop.q,create_nested_type.q,create_or_replace_view.q,create_struct_table.q,create_union_table.q,database.q,database_location.q,database_properties.q,ddltime.q,describe_database_json.q,drop_database_removes_partition_dirs.q,escape1.q,escape2.q,exim_00_nonpart_empty.q,exim_01_nonpart.q,exim_02_00_part_empty.q,exim_02_part.q,exim_03_nonpart_over_compat.q,exim_04_all_part.q,exim_04_evolved_parts.q,exim_05_some_part.q,exim_06_one_part.q,exim_07_all_part_over_nonoverlap.q,exim_08_nonpart_rename.q,
exim_09_part_spec_nonoverlap.q,exim_10_external_managed.q,exim_11_managed_external.q,exim_12_external_location.q,exim_13_managed_location.q,exim_14_managed_location_over_existing.q,exim_15_external_part.q,exim_16_part_external.q,exim_17_part_managed.q,exim_18_part_external.q,exim_19_00_part_external_location.q,exim_19_part_external_location.q,exim_20_part_managed_location.q,exim_21_export_authsuccess.q,exim_22_import_exist_authsuccess.q,exim_23_import_part_authsuccess.q,exim_24_import_nonexist_authsuccess.q,global_limit.q,groupby_complex_types.q,groupby_complex_types_multi_single_reducer.q,index_auth.q,index_auto.q,index_auto_empty.q,index_bitmap.q,index_bitmap1.q,index_bitmap2.q,index_bitmap3.q,index_bitmap_auto.q,index_bitmap_rc.q,index_compact.q,index_compact_1.q,index_compact_2.q,index_compact_3.q,index_stale_partitioned.q,init_file.q,input16.q,input16_cc.q,input46.q,input_columnarserde.q,input_dynamicserde.q,input_lazyserde.q,input_testxpath3.q,input_testxpath4.q,insert2_overwr
ite_partitions.q,insertexternal1.q,join_thrift.q,lateral_view.q,load_binary_data.q,load_exist_part_authsuccess.q,load_nonpart_authsuccess.q,load_part_authsuccess.q,loadpart_err.q,lock1.q,lock2.q,lock3.q,lock4.q,merge_dynamic_partition.q,multi_insert.q,multi_insert_move_tasks_share_dependencies.q,null_column.q,ppd_clusterby.q,query_with_semi.q,rename_column.q,sample6.q,sample_islocalmode_hook.q,set_processor_namespaces.q,show_tables.q,source.q,split_sample.q,str_to_map.q,transform1.q,udaf_collect_set.q,udaf_context_ngrams.q,udaf_histogram_numeric.q,udaf_ngrams.q,udaf_percentile_approx.q,udf_array.q,udf_bitmap_and.q,udf_bitmap_or.q,udf_explode.q,udf_format_number.q,udf_map.q,udf_map_keys.q,udf_map_values.q,udf_max.q,udf_min.q,udf_named_struct.q,udf_percentile.q,udf_printf.q,udf_sentences.q,udf_sort_array.q,udf_split.q,udf_struct.q,udf_substr.q,udf_translate.q,udf_union.q,udf_xpath.q,udtf_stack.q,view.q,virtual_column.q
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MetadataOnlyOptimizer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MetadataOnlyOptimizer.java?rev=1611924&r1=1611923&r2=1611924&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MetadataOnlyOptimizer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MetadataOnlyOptimizer.java Sat Jul 19 17:17:42 2014
@@ -17,11 +17,8 @@
*/
package org.apache.hadoop.hive.ql.optimizer.physical;
-import java.io.Serializable;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -29,33 +26,19 @@ 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.ql.exec.FileSinkOperator;
import org.apache.hadoop.hive.ql.exec.GroupByOperator;
-import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
-import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat;
-import org.apache.hadoop.hive.ql.io.OneNullRowInputFormat;
import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
-import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.GraphWalker;
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.lib.PreOrderWalker;
import org.apache.hadoop.hive.ql.lib.Rule;
import org.apache.hadoop.hive.ql.lib.RuleRegExp;
-import org.apache.hadoop.hive.ql.parse.ParseContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
-import org.apache.hadoop.hive.ql.plan.MapWork;
-import org.apache.hadoop.hive.ql.plan.MapredWork;
-import org.apache.hadoop.hive.ql.plan.OperatorDesc;
-import org.apache.hadoop.hive.ql.plan.PartitionDesc;
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
-import org.apache.hadoop.hive.serde.serdeConstants;
-import org.apache.hadoop.hive.serde2.NullStructSerDe;
/**
*
@@ -72,9 +55,9 @@ import org.apache.hadoop.hive.serde2.Nul
*
*/
public class MetadataOnlyOptimizer implements PhysicalPlanResolver {
- private static final Log LOG = LogFactory.getLog(MetadataOnlyOptimizer.class.getName());
+ static final Log LOG = LogFactory.getLog(MetadataOnlyOptimizer.class.getName());
- static private class WalkerCtx implements NodeProcessorCtx {
+ static class WalkerCtx implements NodeProcessorCtx {
/* operators for which there is chance the optimization can be applied */
private final HashSet<TableScanOperator> possible = new HashSet<TableScanOperator>();
/* operators for which the optimization will be successful */
@@ -129,7 +112,6 @@ public class MetadataOnlyOptimizer imple
@Override
public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
Object... nodeOutputs) throws SemanticException {
- TableScanOperator node = (TableScanOperator) nd;
TableScanOperator tsOp = (TableScanOperator) nd;
WalkerCtx walkerCtx = (WalkerCtx) procCtx;
List<Integer> colIDs = tsOp.getNeededColumnIDs();
@@ -174,151 +156,17 @@ public class MetadataOnlyOptimizer imple
@Override
public PhysicalContext resolve(PhysicalContext pctx) throws SemanticException {
- Dispatcher disp = new MetadataOnlyTaskDispatcher(pctx);
+ Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
+ opRules.put(new RuleRegExp("R1", TableScanOperator.getOperatorName() + "%"),
+ new TableScanProcessor());
+ opRules.put(new RuleRegExp("R2",
+ GroupByOperator.getOperatorName() + "%.*" + FileSinkOperator.getOperatorName() + "%"),
+ new FileSinkProcessor());
+ Dispatcher disp = new NullScanTaskDispatcher(pctx, opRules);
GraphWalker ogw = new DefaultGraphWalker(disp);
ArrayList<Node> topNodes = new ArrayList<Node>();
topNodes.addAll(pctx.getRootTasks());
ogw.startWalking(topNodes, null);
return pctx;
}
-
- /**
- * Iterate over all tasks one-to-one and convert them to metadata only
- */
- class MetadataOnlyTaskDispatcher implements Dispatcher {
-
- private final PhysicalContext physicalContext;
-
- public MetadataOnlyTaskDispatcher(PhysicalContext context) {
- super();
- physicalContext = context;
- }
-
- private String getAliasForTableScanOperator(MapWork work,
- TableScanOperator tso) {
-
- for (Map.Entry<String, Operator<? extends OperatorDesc>> entry :
- work.getAliasToWork().entrySet()) {
- if (entry.getValue() == tso) {
- return entry.getKey();
- }
- }
-
- return null;
- }
-
- private PartitionDesc changePartitionToMetadataOnly(PartitionDesc desc) {
- if (desc != null) {
- desc.setInputFileFormatClass(OneNullRowInputFormat.class);
- desc.setOutputFileFormatClass(HiveIgnoreKeyTextOutputFormat.class);
- desc.getProperties().setProperty(serdeConstants.SERIALIZATION_LIB,
- NullStructSerDe.class.getName());
- }
- return desc;
- }
-
- private List<String> getPathsForAlias(MapWork work, String alias) {
- List<String> paths = new ArrayList<String>();
-
- for (Map.Entry<String, ArrayList<String>> entry : work.getPathToAliases().entrySet()) {
- if (entry.getValue().contains(alias)) {
- paths.add(entry.getKey());
- }
- }
-
- return paths;
- }
-
- private void processAlias(MapWork work, String alias) {
- work.setUseOneNullRowInputFormat(true);
-
- // Change the alias partition desc
- PartitionDesc aliasPartn = work.getAliasToPartnInfo().get(alias);
- changePartitionToMetadataOnly(aliasPartn);
-
- List<String> paths = getPathsForAlias(work, alias);
- for (String path : paths) {
- PartitionDesc partDesc = work.getPathToPartitionInfo().get(path);
- PartitionDesc newPartition = changePartitionToMetadataOnly(partDesc);
- Path fakePath = new Path(physicalContext.getContext().getMRTmpPath()
- + newPartition.getTableName()
- + encode(newPartition.getPartSpec()));
- work.getPathToPartitionInfo().remove(path);
- work.getPathToPartitionInfo().put(fakePath.getName(), newPartition);
- ArrayList<String> aliases = work.getPathToAliases().remove(path);
- work.getPathToAliases().put(fakePath.getName(), aliases);
- }
- }
-
- // considered using URLEncoder, but it seemed too much
- private String encode(Map<String, String> partSpec) {
- return partSpec.toString().replaceAll("[:/#\\?]", "_");
- }
-
- @Override
- public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs)
- throws SemanticException {
- Task<? extends Serializable> task = (Task<? extends Serializable>) nd;
-
- // create a the context for walking operators
- ParseContext parseContext = physicalContext.getParseContext();
- WalkerCtx walkerCtx = new WalkerCtx();
-
- Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
- opRules.put(new RuleRegExp("R1",
- TableScanOperator.getOperatorName() + "%"),
- new TableScanProcessor());
- opRules.put(new RuleRegExp("R2",
- GroupByOperator.getOperatorName() + "%.*" + FileSinkOperator.getOperatorName() + "%"),
- new FileSinkProcessor());
-
- for (MapWork mapWork: task.getMapWork()) {
- LOG.debug("Looking at: "+mapWork.getName());
- Collection<Operator<? extends OperatorDesc>> topOperators
- = mapWork.getAliasToWork().values();
- if (topOperators.size() == 0) {
- LOG.debug("No top operators");
- return null;
- }
-
- LOG.info("Looking for table scans where optimization is applicable");
-
- // The dispatcher fires the processor corresponding to the closest
- // matching rule and passes the context along
- Dispatcher disp = new DefaultRuleDispatcher(null, opRules, walkerCtx);
- GraphWalker ogw = new PreOrderWalker(disp);
-
- // Create a list of topOp nodes
- ArrayList<Node> topNodes = new ArrayList<Node>();
- // Get the top Nodes for this map-reduce task
- for (Operator<? extends OperatorDesc>
- workOperator : topOperators) {
- if (parseContext.getTopOps().values().contains(workOperator)) {
- topNodes.add(workOperator);
- }
- }
-
- Operator<? extends OperatorDesc> reducer = task.getReducer(mapWork);
- if (reducer != null) {
- topNodes.add(reducer);
- }
-
- ogw.startWalking(topNodes, null);
-
- LOG.info(String.format("Found %d metadata only table scans",
- walkerCtx.getMetadataOnlyTableScans().size()));
- Iterator<TableScanOperator> iterator
- = walkerCtx.getMetadataOnlyTableScans().iterator();
-
- while (iterator.hasNext()) {
- TableScanOperator tso = iterator.next();
- ((TableScanDesc)tso.getConf()).setIsMetadataOnly(true);
- String alias = getAliasForTableScanOperator(mapWork, tso);
- LOG.info("Metadata only table scan for " + alias);
- processAlias(mapWork, alias);
- }
- }
- return null;
- }
- }
}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanOptimizer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanOptimizer.java?rev=1611924&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanOptimizer.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanOptimizer.java Sat Jul 19 17:17:42 2014
@@ -0,0 +1,130 @@
+/**
+ * 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.physical;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Stack;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.ql.exec.FilterOperator;
+import org.apache.hadoop.hive.ql.exec.LimitOperator;
+import org.apache.hadoop.hive.ql.exec.TableScanOperator;
+import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
+import org.apache.hadoop.hive.ql.lib.Dispatcher;
+import org.apache.hadoop.hive.ql.lib.GraphWalker;
+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.lib.Rule;
+import org.apache.hadoop.hive.ql.lib.RuleRegExp;
+import org.apache.hadoop.hive.ql.optimizer.physical.MetadataOnlyOptimizer.WalkerCtx;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+
+/**
+ * This optimizer attempts following two optimizations:
+ * 1. If it finds TS followed By FIL which has been determined at compile time to evaluate to
+ * zero, it removes all input paths for that table scan.
+ * 2. If it finds TS followed by Limit 0, it removes all input paths from table scan.
+ */
+public class NullScanOptimizer implements PhysicalPlanResolver {
+
+ private static final Log LOG = LogFactory.getLog(NullScanOptimizer.class.getName());
+ @Override
+ public PhysicalContext resolve(PhysicalContext pctx) throws SemanticException {
+
+ Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
+ opRules.put(new RuleRegExp("R1", TableScanOperator.getOperatorName() + "%.*" +
+ FilterOperator.getOperatorName() + "%"), new WhereFalseProcessor());
+ Dispatcher disp = new NullScanTaskDispatcher(pctx, opRules);
+ GraphWalker ogw = new DefaultGraphWalker(disp);
+ ArrayList<Node> topNodes = new ArrayList<Node>();
+ topNodes.addAll(pctx.getRootTasks());
+ ogw.startWalking(topNodes, null);
+
+ opRules.clear();
+
+ opRules.put(new RuleRegExp("R1", TableScanOperator.getOperatorName()+ "%"),new TSMarker());
+ opRules.put(new RuleRegExp("R2", LimitOperator.getOperatorName()+ "%"), new Limit0Processor());
+ disp = new NullScanTaskDispatcher(pctx, opRules);
+ ogw = new DefaultGraphWalker(disp);
+ topNodes = new ArrayList<Node>();
+ topNodes.addAll(pctx.getRootTasks());
+ ogw.startWalking(topNodes, null);
+ return pctx;
+ }
+
+ static private class WhereFalseProcessor implements NodeProcessor {
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ FilterOperator filter = (FilterOperator) nd;
+ ExprNodeDesc condition = filter.getConf().getPredicate();
+ if (!(condition instanceof ExprNodeConstantDesc)) {
+ return null;
+ }
+ ExprNodeConstantDesc c = (ExprNodeConstantDesc) condition;
+ if (c.getValue() != Boolean.FALSE) {
+ return null;
+ }
+
+ WalkerCtx ctx = (WalkerCtx) procCtx;
+ for (Node op : stack) {
+ if (op instanceof TableScanOperator) {
+ ctx.setMayBeMetadataOnly((TableScanOperator)op);
+ LOG.info("Found where false TableScan. " + op);
+ }
+ }
+ ctx.convertMetadataOnly();
+ return null;
+ }
+ }
+
+ static private class Limit0Processor implements NodeProcessor {
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ if(!(((LimitOperator)nd).getConf().getLimit() == 0)) {
+ return null;
+ }
+ LOG.info("Found Limit 0 TableScan. " + nd);
+ ((WalkerCtx)procCtx).convertMetadataOnly();
+ return null;
+ }
+
+ }
+
+ static private class TSMarker implements NodeProcessor {
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ ((WalkerCtx)procCtx).setMayBeMetadataOnly((TableScanOperator)nd);
+ return null;
+ }
+ }
+}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java?rev=1611924&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/NullScanTaskDispatcher.java Sat Jul 19 17:17:42 2014
@@ -0,0 +1,187 @@
+/**
+ * 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.physical;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+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.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.TableScanOperator;
+import org.apache.hadoop.hive.ql.exec.Task;
+import org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat;
+import org.apache.hadoop.hive.ql.io.OneNullRowInputFormat;
+import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
+import org.apache.hadoop.hive.ql.lib.Dispatcher;
+import org.apache.hadoop.hive.ql.lib.GraphWalker;
+import org.apache.hadoop.hive.ql.lib.Node;
+import org.apache.hadoop.hive.ql.lib.NodeProcessor;
+import org.apache.hadoop.hive.ql.lib.PreOrderWalker;
+import org.apache.hadoop.hive.ql.lib.Rule;
+import org.apache.hadoop.hive.ql.optimizer.physical.MetadataOnlyOptimizer.WalkerCtx;
+import org.apache.hadoop.hive.ql.parse.ParseContext;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.MapWork;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.PartitionDesc;
+import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.hive.serde2.NullStructSerDe;
+
+/**
+ * Iterate over all tasks one by one and removes all input paths from task if conditions as
+ * defined in rules match.
+ */
+public class NullScanTaskDispatcher implements Dispatcher {
+
+ private final PhysicalContext physicalContext;
+ private final Map<Rule, NodeProcessor> rules;
+ static final Log LOG = LogFactory.getLog(NullScanTaskDispatcher.class.getName());
+
+ public NullScanTaskDispatcher(PhysicalContext context, Map<Rule, NodeProcessor> rules) {
+ super();
+ physicalContext = context;
+ this.rules = rules;
+ }
+
+ private String getAliasForTableScanOperator(MapWork work,
+ TableScanOperator tso) {
+
+ for (Map.Entry<String, Operator<? extends OperatorDesc>> entry :
+ work.getAliasToWork().entrySet()) {
+ if (entry.getValue() == tso) {
+ return entry.getKey();
+ }
+ }
+
+ return null;
+ }
+
+ private PartitionDesc changePartitionToMetadataOnly(PartitionDesc desc) {
+ if (desc != null) {
+ desc.setInputFileFormatClass(OneNullRowInputFormat.class);
+ desc.setOutputFileFormatClass(HiveIgnoreKeyTextOutputFormat.class);
+ desc.getProperties().setProperty(serdeConstants.SERIALIZATION_LIB,
+ NullStructSerDe.class.getName());
+ }
+ return desc;
+ }
+
+ private List<String> getPathsForAlias(MapWork work, String alias) {
+ List<String> paths = new ArrayList<String>();
+
+ for (Map.Entry<String, ArrayList<String>> entry : work.getPathToAliases().entrySet()) {
+ if (entry.getValue().contains(alias)) {
+ paths.add(entry.getKey());
+ }
+ }
+
+ return paths;
+ }
+
+ private void processAlias(MapWork work, String alias) {
+ work.setUseOneNullRowInputFormat(true);
+
+ // Change the alias partition desc
+ PartitionDesc aliasPartn = work.getAliasToPartnInfo().get(alias);
+ changePartitionToMetadataOnly(aliasPartn);
+
+ List<String> paths = getPathsForAlias(work, alias);
+ for (String path : paths) {
+ PartitionDesc partDesc = work.getPathToPartitionInfo().get(path);
+ PartitionDesc newPartition = changePartitionToMetadataOnly(partDesc);
+ Path fakePath = new Path(physicalContext.getContext().getMRTmpPath()
+ + newPartition.getTableName()
+ + encode(newPartition.getPartSpec()));
+ work.getPathToPartitionInfo().remove(path);
+ work.getPathToPartitionInfo().put(fakePath.getName(), newPartition);
+ ArrayList<String> aliases = work.getPathToAliases().remove(path);
+ work.getPathToAliases().put(fakePath.getName(), aliases);
+ }
+ }
+
+ // considered using URLEncoder, but it seemed too much
+ private String encode(Map<String, String> partSpec) {
+ return partSpec.toString().replaceAll("[:/#\\?]", "_");
+ }
+
+ @Override
+ public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs)
+ throws SemanticException {
+ Task<? extends Serializable> task = (Task<? extends Serializable>) nd;
+
+ // create a the context for walking operators
+ ParseContext parseContext = physicalContext.getParseContext();
+ WalkerCtx walkerCtx = new WalkerCtx();
+
+ for (MapWork mapWork: task.getMapWork()) {
+ LOG.debug("Looking at: "+mapWork.getName());
+ Collection<Operator<? extends OperatorDesc>> topOperators
+ = mapWork.getAliasToWork().values();
+ if (topOperators.size() == 0) {
+ LOG.debug("No top operators");
+ return null;
+ }
+
+ LOG.info("Looking for table scans where optimization is applicable");
+
+ // The dispatcher fires the processor corresponding to the closest
+ // matching rule and passes the context along
+ Dispatcher disp = new DefaultRuleDispatcher(null, rules, walkerCtx);
+ GraphWalker ogw = new PreOrderWalker(disp);
+
+ // Create a list of topOp nodes
+ ArrayList<Node> topNodes = new ArrayList<Node>();
+ // Get the top Nodes for this task
+ for (Operator<? extends OperatorDesc>
+ workOperator : topOperators) {
+ if (parseContext.getTopOps().values().contains(workOperator)) {
+ topNodes.add(workOperator);
+ }
+ }
+
+ Operator<? extends OperatorDesc> reducer = task.getReducer(mapWork);
+ if (reducer != null) {
+ topNodes.add(reducer);
+ }
+
+ ogw.startWalking(topNodes, null);
+
+ LOG.info(String.format("Found %d null table scans",
+ walkerCtx.getMetadataOnlyTableScans().size()));
+ Iterator<TableScanOperator> iterator
+ = walkerCtx.getMetadataOnlyTableScans().iterator();
+
+ while (iterator.hasNext()) {
+ TableScanOperator tso = iterator.next();
+ tso.getConf().setIsMetadataOnly(true);
+ String alias = getAliasForTableScanOperator(mapWork, tso);
+ LOG.info("Null table scan for " + alias);
+ processAlias(mapWork, alias);
+ }
+ }
+ return null;
+ }
+}
\ No newline at end of file
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java?rev=1611924&r1=1611923&r2=1611924&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java Sat Jul 19 17:17:42 2014
@@ -67,6 +67,9 @@ public class PhysicalOptimizer {
if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVEMETADATAONLYQUERIES)) {
resolvers.add(new MetadataOnlyOptimizer());
}
+ if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVENULLSCANOPTIMIZE)) {
+ resolvers.add(new NullScanOptimizer());
+ }
if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVESAMPLINGFORORDERBY)) {
resolvers.add(new SamplingOptimizer());
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java?rev=1611924&r1=1611923&r2=1611924&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/TezCompiler.java Sat Jul 19 17:17:42 2014
@@ -61,6 +61,7 @@ import org.apache.hadoop.hive.ql.optimiz
import org.apache.hadoop.hive.ql.optimizer.SetReducerParallelism;
import org.apache.hadoop.hive.ql.optimizer.physical.CrossProductCheck;
import org.apache.hadoop.hive.ql.optimizer.physical.MetadataOnlyOptimizer;
+import org.apache.hadoop.hive.ql.optimizer.physical.NullScanOptimizer;
import org.apache.hadoop.hive.ql.optimizer.physical.PhysicalContext;
import org.apache.hadoop.hive.ql.optimizer.physical.Vectorizer;
import org.apache.hadoop.hive.ql.optimizer.physical.StageIDsRearranger;
@@ -248,6 +249,12 @@ public class TezCompiler extends TaskCom
PhysicalContext physicalCtx = new PhysicalContext(conf, pCtx, pCtx.getContext(), rootTasks,
pCtx.getFetchTask());
+ if (conf.getBoolVar(HiveConf.ConfVars.HIVENULLSCANOPTIMIZE)) {
+ physicalCtx = new NullScanOptimizer().resolve(physicalCtx);
+ } else {
+ LOG.debug("Skipping null scan query optimization");
+ }
+
if (conf.getBoolVar(HiveConf.ConfVars.HIVEMETADATAONLYQUERIES)) {
physicalCtx = new MetadataOnlyOptimizer().resolve(physicalCtx);
} else {
Added: hive/trunk/ql/src/test/queries/clientpositive/optimize_nullscan.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/optimize_nullscan.q?rev=1611924&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/optimize_nullscan.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/optimize_nullscan.q Sat Jul 19 17:17:42 2014
@@ -0,0 +1,19 @@
+explain extended
+select key from src where false;
+select key from src where false;
+
+explain extended
+select * from (select key from src where false) a left outer join (select key from srcpart limit 0) b on a.key=b.key;
+select * from (select key from src where false) a left outer join (select key from srcpart limit 0) b on a.key=b.key;
+
+explain extended
+select count(key) from src where false union all select count(key) from srcpart ;
+select count(key) from src where false union all select count(key) from srcpart ;
+
+explain extended
+select * from (select key from src where false) a left outer join (select value from srcpart limit 0) b ;
+select * from (select key from src where false) a left outer join (select value from srcpart limit 0) b ;
+
+explain extended
+select * from (select key from src union all select src.key from src left outer join srcpart on src.key = srcpart.key) a where false;
+select * from (select key from src union all select src.key from src left outer join srcpart on src.key = srcpart.key) a where false;
Added: hive/trunk/ql/src/test/results/clientpositive/optimize_nullscan.q.out
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/results/clientpositive/optimize_nullscan.q.out?rev=1611924&view=auto
==============================================================================
Files hive/trunk/ql/src/test/results/clientpositive/optimize_nullscan.q.out (added) and hive/trunk/ql/src/test/results/clientpositive/optimize_nullscan.q.out Sat Jul 19 17:17:42 2014 differ
Added: hive/trunk/ql/src/test/results/clientpositive/tez/optimize_nullscan.q.out
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/results/clientpositive/tez/optimize_nullscan.q.out?rev=1611924&view=auto
==============================================================================
Files hive/trunk/ql/src/test/results/clientpositive/tez/optimize_nullscan.q.out (added) and hive/trunk/ql/src/test/results/clientpositive/tez/optimize_nullscan.q.out Sat Jul 19 17:17:42 2014 differ