You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by na...@apache.org on 2010/04/07 01:04:12 UTC
svn commit: r931363 [2/27] - in /hadoop/hive/trunk: ./
contrib/src/test/results/clientpositive/
ql/src/java/org/apache/hadoop/hive/ql/
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/hooks/ ql/src/java/org/apache/hadoo...
Added: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/LineageCtx.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/LineageCtx.java?rev=931363&view=auto
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/LineageCtx.java (added)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/LineageCtx.java Tue Apr 6 23:04:06 2010
@@ -0,0 +1,192 @@
+/**
+ * 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.lineage;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.BaseColumnInfo;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.Dependency;
+import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.parse.ParseContext;
+
+/**
+ * This class contains the lineage context that is passed
+ * while walking the operator tree in Lineage. The context
+ * contains the LineageInfo structure that is passed to the
+ * pre-execution hooks.
+ */
+public class LineageCtx implements NodeProcessorCtx {
+
+ public static class Index {
+
+ /**
+ * Serial Version UID.
+ */
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * The map contains an index from the (operator, columnInfo) to the
+ * dependency vector for that tuple. This is used to generate the
+ * dependency vectors during the walk of the operator tree.
+ */
+ private final Map<Operator<? extends Serializable>, LinkedHashMap<ColumnInfo, Dependency>> depMap;
+
+ /**
+ * Constructor.
+ */
+ public Index() {
+ depMap = new LinkedHashMap<Operator<? extends Serializable>, LinkedHashMap<ColumnInfo, Dependency>>();
+ }
+
+ /**
+ * Gets the dependency for an operator, columninfo tuple.
+ * @param op The operator whose dependency is being inspected.
+ * @param col The column info whose dependency is being inspected.
+ * @return Dependency for that particular operator, columninfo tuple.
+ * null if no dependency is found.
+ */
+ public Dependency getDependency(Operator<? extends Serializable> op, ColumnInfo col) {
+ Map<ColumnInfo, Dependency> colMap = depMap.get(op);
+ if (colMap == null) {
+ return null;
+ }
+
+ return colMap.get(col);
+ }
+
+ /**
+ * Puts the dependency for an operator, columninfo tuple.
+ * @param op The operator whose dependency is being inserted.
+ * @param col The column info whose dependency is being inserted.
+ * @param dep The dependency.
+ */
+ public void putDependency(Operator<? extends Serializable> op,
+ ColumnInfo col, Dependency dep) {
+ LinkedHashMap<ColumnInfo, Dependency> colMap = depMap.get(op);
+ if (colMap == null) {
+ colMap = new LinkedHashMap<ColumnInfo, Dependency>();
+ depMap.put(op, colMap);
+ }
+ colMap.put(col, dep);
+ }
+
+ /**
+ * Merges the new dependencies in dep to the existing dependencies
+ * of (op, ci).
+ *
+ * @param op The operator of the column whose dependency is being modified.
+ * @param ci The column info of the associated column.
+ * @param dependency The new dependency.
+ */
+ public void mergeDependency(Operator<? extends Serializable> op,
+ ColumnInfo ci, Dependency dep) {
+ Dependency old_dep = getDependency(op, ci);
+ if (old_dep == null) {
+ putDependency(op, ci, dep);
+ } else {
+ LineageInfo.DependencyType new_type =
+ LineageCtx.getNewDependencyType(old_dep.getType(),
+ LineageInfo.DependencyType.EXPRESSION);
+ old_dep.setType(new_type);
+ Set<BaseColumnInfo> bci_set = new LinkedHashSet<BaseColumnInfo>(old_dep.getBaseCols());
+ bci_set.addAll(dep.getBaseCols());
+ old_dep.setBaseCols(new ArrayList<BaseColumnInfo>(bci_set));
+ // TODO: Fix the expressions later.
+ old_dep.setExpr(null);
+ }
+ }
+ }
+
+ /**
+ * The map contains an index from the (operator, columnInfo) to the
+ * dependency vector for that tuple. This is used to generate the
+ * dependency vectors during the walk of the operator tree.
+ */
+ private final Index index;
+
+ /**
+ * Parse context to get to the table metadata information.
+ */
+ private final ParseContext pctx;
+
+ /**
+ * Constructor.
+ *
+ * @param pctx The parse context that is used to get table metadata information.
+ */
+ public LineageCtx(ParseContext pctx) {
+ index = new Index();
+ this.pctx = pctx;
+ }
+
+ /**
+ * Gets the parse context.
+ *
+ * @return ParseContext
+ */
+ public ParseContext getParseCtx() {
+ return pctx;
+ }
+
+ /**
+ * Gets the dependency index.
+ *
+ * @return Index
+ */
+ public Index getIndex() {
+ return index;
+ }
+
+ /**
+ * Gets the new dependency type by comparing the old dependency type and the
+ * current dependency type. The current dependency type is the dependency imposed
+ * by the current expression. Typically the dependency type is computed using
+ * the following rules:
+ * SCRIPT - In case anywhere in the lineage tree there was a script operator, otherwise
+ * EXPRESSION - In case anywhere in the lineage tree a union,
+ * udf, udaf or udtf was done, otherwise
+ * SIMPLE - This captures direct column copies.
+ *
+ * @param old_type The old dependency type.
+ * @param curr_type The current operators dependency type.
+ * @return
+ */
+ public static LineageInfo.DependencyType getNewDependencyType(
+ LineageInfo.DependencyType old_type, LineageInfo.DependencyType curr_type) {
+ if (old_type == LineageInfo.DependencyType.SCRIPT ||
+ curr_type == LineageInfo.DependencyType.SCRIPT) {
+ return LineageInfo.DependencyType.SCRIPT;
+ }
+
+ if (old_type == LineageInfo.DependencyType.EXPRESSION ||
+ curr_type == LineageInfo.DependencyType.EXPRESSION) {
+ return LineageInfo.DependencyType.EXPRESSION;
+ }
+
+ return LineageInfo.DependencyType.SIMPLE;
+ }
+}
Added: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java?rev=931363&view=auto
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java (added)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/lineage/OpProcFactory.java Tue Apr 6 23:04:06 2010
@@ -0,0 +1,519 @@
+/**
+ * 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.lineage;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+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.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.exec.ForwardOperator;
+import org.apache.hadoop.hive.ql.exec.GroupByOperator;
+import org.apache.hadoop.hive.ql.exec.JoinOperator;
+import org.apache.hadoop.hive.ql.exec.LateralViewJoinOperator;
+import org.apache.hadoop.hive.ql.exec.Operator;
+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.hooks.LineageInfo;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.BaseColumnInfo;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.Dependency;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.DependencyType;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.TableAliasInfo;
+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.Utils;
+import org.apache.hadoop.hive.ql.parse.ParseContext;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.AggregationDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.JoinDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
+
+/**
+ * Operator factory for the rule processors for lineage.
+ */
+public class OpProcFactory {
+
+ /**
+ * Returns the parent operator in the walk path to the current operator.
+ *
+ * @param stack The stack encoding the path.
+ *
+ * @return Operator The parent operator in the current path.
+ */
+ protected static Operator<? extends Serializable> getParent(Stack<Node> stack) {
+ return (Operator<? extends Serializable>)Utils.getNthAncestor(stack, 1);
+ }
+
+ /**
+ * Processor for Script and UDTF Operators.
+ */
+ public static class TransformLineage extends DefaultLineage implements NodeProcessor {
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ // LineageCTx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+
+ // The operators
+ Operator<? extends Serializable> op = (Operator<? extends Serializable>)nd;
+ Operator<? extends Serializable> inpOp = getParent(stack);
+
+ // Create a single dependency list by concatenating the dependencies of all
+ // the cols
+ Dependency dep = new Dependency();
+ DependencyType new_type = LineageInfo.DependencyType.SCRIPT;
+ dep.setType(LineageInfo.DependencyType.SCRIPT);
+ // TODO: Fix this to a non null value.
+ dep.setExpr(null);
+
+ LinkedHashSet<BaseColumnInfo> col_set = new LinkedHashSet<BaseColumnInfo>();
+ for(ColumnInfo ci : inpOp.getSchema().getSignature()) {
+ Dependency d = lCtx.getIndex().getDependency(inpOp, ci);
+ if (d != null) {
+ new_type = LineageCtx.getNewDependencyType(d.getType(), new_type);
+ col_set.addAll(d.getBaseCols());
+ }
+ }
+
+ dep.setType(new_type);
+ dep.setBaseCols(new ArrayList<BaseColumnInfo>(col_set));
+
+ // This dependency is then set for all the colinfos of the script operator
+ for(ColumnInfo ci : op.getSchema().getSignature()) {
+ lCtx.getIndex().putDependency(op, ci, dep);
+ }
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Processor for TableScan Operator. This actually creates the base column mappings.
+ */
+ public static class TableScanLineage extends DefaultLineage implements NodeProcessor {
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ // LineageCtx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+ ParseContext pctx = lCtx.getParseCtx();
+
+ // Table scan operator.
+ TableScanOperator top = (TableScanOperator)nd;
+ org.apache.hadoop.hive.ql.metadata.Table t = pctx.getTopToTable().get(top);
+ Table tab = t.getTTable();
+
+ // Generate the mappings
+ RowSchema rs = top.getSchema();
+ List<FieldSchema> cols = t.getAllCols();
+ TableAliasInfo tai = new TableAliasInfo();
+ tai.setAlias(top.getConf().getAlias());
+ tai.setTable(tab);
+ int cnt = 0;
+ for(ColumnInfo ci : rs.getSignature()) {
+ // Create a dependency
+ Dependency dep = new Dependency();
+ BaseColumnInfo bci = new BaseColumnInfo();
+ bci.setTabAlias(tai);
+ bci.setColumn(cols.get(cnt++));
+
+ // Populate the dependency
+ dep.setType(LineageInfo.DependencyType.SIMPLE);
+ // TODO: Find out how to get the expression here.
+ dep.setExpr(null);
+ dep.setBaseCols(new ArrayList<BaseColumnInfo>());
+ dep.getBaseCols().add(bci);
+
+ // Put the dependency in the map
+ lCtx.getIndex().putDependency(top, ci, dep);
+ }
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Processor for Join Operator.
+ */
+ public static class JoinLineage extends DefaultLineage implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ // Assert that there is atleast one item in the stack. This should never
+ // be called for leafs.
+ assert(!stack.isEmpty());
+
+ // LineageCtx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+ JoinOperator op = (JoinOperator)nd;
+ JoinDesc jd = op.getConf();
+
+ // The input operator to the join is always a reduce sink operator
+ ReduceSinkOperator inpOp = (ReduceSinkOperator)getParent(stack);
+ ReduceSinkDesc rd = inpOp.getConf();
+ int tag = rd.getTag();
+
+ // Iterate over the outputs of the join operator and merge the
+ // dependencies of the columns that corresponding to the tag.
+ int cnt = 0;
+ List<ExprNodeDesc> exprs = jd.getExprs().get((byte)tag);
+ for(ColumnInfo ci : op.getSchema().getSignature()) {
+ if (jd.getReversedExprs().get(ci.getInternalName()) != tag) {
+ continue;
+ }
+
+ // Otherwise look up the expression corresponding to this ci
+ ExprNodeDesc expr = exprs.get(cnt++);
+ lCtx.getIndex().mergeDependency(op, ci,
+ ExprProcFactory.getExprDependency(lCtx, inpOp, expr));
+ }
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Processor for Join Operator.
+ */
+ public static class LateralViewJoinLineage extends DefaultLineage implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ // Assert that there is atleast one item in the stack. This should never
+ // be called for leafs.
+ assert(!stack.isEmpty());
+
+ // LineageCtx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+ LateralViewJoinOperator op = (LateralViewJoinOperator)nd;
+ boolean isUdtfPath = true;
+ Operator<? extends Serializable> inpOp = getParent(stack);
+ ArrayList<ColumnInfo> cols = inpOp.getSchema().getSignature();
+
+ if (inpOp instanceof SelectOperator) {
+ isUdtfPath = false;
+ }
+
+ // Dirty hack!!
+ // For the select path the columns are the ones at the end of the
+ // current operators schema and for the udtf path the columns are
+ // at the beginning of the operator schema.
+ ArrayList<ColumnInfo> out_cols = op.getSchema().getSignature();
+ int out_cols_size = out_cols.size();
+ int cols_size = cols.size();
+ if (isUdtfPath) {
+ int cnt = 0;
+ while (cnt < cols_size) {
+ lCtx.getIndex().mergeDependency(op, out_cols.get(cnt),
+ lCtx.getIndex().getDependency(inpOp, cols.get(cnt)));
+ cnt++;
+ }
+ }
+ else {
+ int cnt = cols_size - 1;
+ while (cnt >= 0) {
+ lCtx.getIndex().mergeDependency(op, out_cols.get(out_cols_size - cols_size + cnt),
+ lCtx.getIndex().getDependency(inpOp, cols.get(cnt)));
+ cnt--;
+ }
+ }
+ return null;
+ }
+
+ }
+
+ /**
+ * Processor for Select operator.
+ */
+ public static class SelectLineage extends DefaultLineage implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ LineageCtx lctx = (LineageCtx)procCtx;
+ SelectOperator sop = (SelectOperator)nd;
+
+ // if this is a selStarNoCompute then this select operator
+ // is treated like a default operator, so just call the super classes
+ // process method.
+ if (sop.getConf().isSelStarNoCompute()) {
+ return super.process(nd, stack, procCtx, nodeOutputs);
+ }
+
+ // Otherwise we treat this as a normal select operator and look at
+ // the expressions.
+
+ ArrayList<ColumnInfo> col_infos = sop.getSchema().getSignature();
+ int cnt = 0;
+ for(ExprNodeDesc expr : sop.getConf().getColList()) {
+ lctx.getIndex().putDependency(sop, col_infos.get(cnt++),
+ ExprProcFactory.getExprDependency(lctx, getParent(stack), expr));
+ }
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Processor for GroupBy operator.
+ */
+ public static class GroupByLineage extends DefaultLineage implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ LineageCtx lctx = (LineageCtx)procCtx;
+ GroupByOperator gop = (GroupByOperator)nd;
+ ArrayList<ColumnInfo> col_infos = gop.getSchema().getSignature();
+ Operator<? extends Serializable> inpOp = getParent(stack);
+ int cnt = 0;
+
+ for(ExprNodeDesc expr : gop.getConf().getKeys()) {
+ lctx.getIndex().putDependency(gop, col_infos.get(cnt++),
+ ExprProcFactory.getExprDependency(lctx, inpOp, expr));
+ }
+
+ for(AggregationDesc agg : gop.getConf().getAggregators()) {
+ // Concatenate the dependencies of all the parameters to
+ // create the new dependency
+ Dependency dep = new Dependency();
+ DependencyType new_type = LineageInfo.DependencyType.EXPRESSION;
+ // TODO: Get the actual string here.
+ dep.setExpr(null);
+ LinkedHashSet<BaseColumnInfo> bci_set = new LinkedHashSet<BaseColumnInfo>();
+ for(ExprNodeDesc expr : agg.getParameters()) {
+ Dependency expr_dep = ExprProcFactory.getExprDependency(lctx, inpOp, expr);
+ if (expr_dep != null) {
+ new_type = LineageCtx.getNewDependencyType(expr_dep.getType(), new_type);
+ bci_set.addAll(expr_dep.getBaseCols());
+ }
+ }
+
+ // If the bci_set is empty, this means that the inputs to this
+ // aggregate function were all constants (e.g. count(1)). In this case
+ // the aggregate function is just dependent on all the tables that are in
+ // the dependency list of the input operator.
+ if (bci_set.isEmpty()) {
+ Set<TableAliasInfo> tai_set = new LinkedHashSet<TableAliasInfo>();
+ for(ColumnInfo ci : inpOp.getSchema().getSignature()) {
+ Dependency inp_dep = lctx.getIndex().getDependency(inpOp, ci);
+ // The dependency can be null as some of the input cis may not have
+ // been set in case of joins.
+ if (inp_dep != null) {
+ for(BaseColumnInfo bci : inp_dep.getBaseCols()) {
+ new_type = LineageCtx.getNewDependencyType(inp_dep.getType(), new_type);
+ tai_set.add(bci.getTabAlias());
+ }
+ }
+ }
+
+ // Create the BaseColumnInfos and set them in the bci_set
+ for(TableAliasInfo tai : tai_set) {
+ BaseColumnInfo bci = new BaseColumnInfo();
+ bci.setTabAlias(tai);
+ // This is set to null to reflect that the dependency is not on any
+ // particular column of the table.
+ bci.setColumn(null);
+ bci_set.add(bci);
+ }
+ }
+
+ dep.setBaseCols(new ArrayList<BaseColumnInfo>(bci_set));
+ dep.setType(new_type);
+ lctx.getIndex().putDependency(gop, col_infos.get(cnt++), dep);
+ }
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Union processor.
+ * In this case we call mergeDependency as opposed to putDependency
+ * in order to account for visits from different parents.
+ */
+ public static class UnionLineage extends DefaultLineage implements NodeProcessor {
+
+ protected static final Log LOG = LogFactory.getLog(OpProcFactory.class.getName());
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ // Assert that there is atleast one item in the stack. This should never
+ // be called for leafs.
+ assert(!stack.isEmpty());
+
+ // LineageCtx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+ Operator<? extends Serializable> op = (Operator<? extends Serializable>)nd;
+
+ // Get the row schema of the input operator.
+ // The row schema of the parent operator
+ Operator<? extends Serializable> inpOp = getParent(stack);
+ RowSchema rs = op.getSchema();
+ ArrayList<ColumnInfo> inp_cols = inpOp.getSchema().getSignature();
+ int cnt = 0;
+ for(ColumnInfo ci : rs.getSignature()) {
+ lCtx.getIndex().mergeDependency(op, ci,
+ lCtx.getIndex().getDependency(inpOp, inp_cols.get(cnt++)));
+ }
+ return null;
+ }
+ }
+
+ /**
+ * ReduceSink processor.
+ */
+ public static class ReduceSinkLineage implements NodeProcessor {
+
+ protected static final Log LOG = LogFactory.getLog(OpProcFactory.class.getName());
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ // Assert that there is atleast one item in the stack. This should never
+ // be called for leafs.
+ assert(!stack.isEmpty());
+
+ // LineageCtx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+ ReduceSinkOperator rop = (ReduceSinkOperator)nd;
+
+ ArrayList<ColumnInfo> col_infos = rop.getSchema().getSignature();
+ Operator<? extends Serializable> inpOp = getParent(stack);
+ int cnt = 0;
+
+ // The keys are included only in case the reduce sink feeds into
+ // a group by operator through a chain of forward operators
+ Operator<? extends Serializable> op = rop.getChildOperators().get(0);
+ while (op instanceof ForwardOperator) {
+ op = op.getChildOperators().get(0);
+ }
+
+ if (op instanceof GroupByOperator) {
+ for(ExprNodeDesc expr : rop.getConf().getKeyCols()) {
+ lCtx.getIndex().putDependency(rop, col_infos.get(cnt++),
+ ExprProcFactory.getExprDependency(lCtx, inpOp, expr));
+ }
+ }
+
+ for(ExprNodeDesc expr : rop.getConf().getValueCols()) {
+ lCtx.getIndex().putDependency(rop, col_infos.get(cnt++),
+ ExprProcFactory.getExprDependency(lCtx, inpOp, expr));
+ }
+
+ return null;
+ }
+ }
+
+ /**
+ * Default processor. This basically passes the input dependencies as such
+ * to the output dependencies.
+ */
+ public static class DefaultLineage implements NodeProcessor {
+
+ protected static final Log LOG = LogFactory.getLog(OpProcFactory.class.getName());
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ // Assert that there is atleast one item in the stack. This should never
+ // be called for leafs.
+ assert(!stack.isEmpty());
+
+ // LineageCtx
+ LineageCtx lCtx = (LineageCtx) procCtx;
+ Operator<? extends Serializable> op = (Operator<? extends Serializable>)nd;
+
+ // Get the row schema of the input operator.
+ // The row schema of the parent operator
+ Operator<? extends Serializable> inpOp = getParent(stack);
+ RowSchema rs = op.getSchema();
+ ArrayList<ColumnInfo> inp_cols = inpOp.getSchema().getSignature();
+ int cnt = 0;
+ for(ColumnInfo ci : rs.getSignature()) {
+ lCtx.getIndex().putDependency(op, ci,
+ lCtx.getIndex().getDependency(inpOp, inp_cols.get(cnt++)));
+ }
+ return null;
+ }
+ }
+
+ public static NodeProcessor getJoinProc() {
+ return new JoinLineage();
+ }
+
+ public static NodeProcessor getLateralViewJoinProc() {
+ return new LateralViewJoinLineage();
+ }
+
+ public static NodeProcessor getTSProc() {
+ return new TableScanLineage();
+ }
+
+ public static NodeProcessor getTransformProc() {
+ return new TransformLineage();
+ }
+
+ public static NodeProcessor getSelProc() {
+ return new SelectLineage();
+ }
+
+ public static NodeProcessor getGroupByProc() {
+ return new GroupByLineage();
+ }
+
+ public static NodeProcessor getUnionProc() {
+ return new UnionLineage();
+ }
+
+ public static NodeProcessor getReduceSinkProc() {
+ return new ReduceSinkLineage();
+ }
+
+ public static NodeProcessor getDefaultProc() {
+ return new DefaultLineage();
+ }
+
+}
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java Tue Apr 6 23:04:06 2010
@@ -37,6 +37,7 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.ql.Context;
import org.apache.hadoop.hive.ql.exec.FetchTask;
import org.apache.hadoop.hive.ql.exec.Task;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo;
import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.io.IgnoreKeyTextOutputFormat;
@@ -58,14 +59,6 @@ import org.apache.hadoop.mapred.TextInpu
* BaseSemanticAnalyzer.
*
*/
-/**
- * BaseSemanticAnalyzer.
- *
- */
-/**
- * BaseSemanticAnalyzer.
- *
- */
public abstract class BaseSemanticAnalyzer {
protected final Hive db;
protected final HiveConf conf;
@@ -88,6 +81,10 @@ public abstract class BaseSemanticAnalyz
* List of WriteEntities that are passed to the hooks.
*/
protected HashSet<WriteEntity> outputs;
+ /**
+ * Lineage information for the query.
+ */
+ protected LineageInfo linfo;
protected static final String TEXTFILE_INPUT = TextInputFormat.class
.getName();
@@ -502,4 +499,22 @@ public abstract class BaseSemanticAnalyz
}
}
}
+
+ /**
+ * Gets the lineage information.
+ *
+ * @return LineageInfo associated with the query.
+ */
+ public LineageInfo getLineageInfo() {
+ return linfo;
+ }
+
+ /**
+ * Sets the lineage information.
+ *
+ * @param linfo The LineageInfo structure that is set in the optimization phase.
+ */
+ public void setLineageInfo(LineageInfo linfo) {
+ this.linfo = linfo;
+ }
}
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseContext.java Tue Apr 6 23:04:06 2010
@@ -33,6 +33,7 @@ import org.apache.hadoop.hive.ql.exec.Jo
import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.optimizer.unionproc.UnionProcContext;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
@@ -48,7 +49,7 @@ import org.apache.hadoop.hive.ql.plan.Fi
* populated. Note that since the parse context contains the operator tree, it
* can be easily retrieved by the next optimization step or finally for task
* generation after the plan has been completely optimized.
- *
+ *
**/
public class ParseContext {
@@ -75,6 +76,11 @@ public class ParseContext {
private Map<GroupByOperator, Set<String>> groupOpToInputTables;
private Map<String, PrunedPartitionList> prunedPartitions;
+ /**
+ * The lineage information.
+ */
+ private LineageInfo lInfo;
+
// is set to true if the expression only contains partitioning columns and not
// any other column reference.
// This is used to optimize select * from table where ... scenario, when the
@@ -105,6 +111,7 @@ public class ParseContext {
* context needed join processing (map join specifically)
* @param topToTable
* the top tables being processed
+ * @param fopToTable the table schemas that are being inserted into
* @param loadTableWork
* list of destination tables being loaded
* @param loadFileWork
@@ -383,7 +390,7 @@ public class ParseContext {
/**
* Sets the hasNonPartCols flag.
- *
+ *
* @param val
*/
public void setHasNonPartCols(boolean val) {
@@ -443,6 +450,24 @@ public class ParseContext {
this.prunedPartitions = prunedPartitions;
}
+ /**
+ * Sets the lineage information.
+ *
+ * @param lInfo The lineage information.
+ */
+ public void setLineageInfo(LineageInfo lInfo) {
+ this.lInfo = lInfo;
+ }
+
+ /**
+ * Gets the associated lineage information.
+ *
+ * @return LineageInfo
+ */
+ public LineageInfo getLineageInfo() {
+ return lInfo;
+ }
+
public Map<MapJoinOperator, QBJoinTree> getMapJoinContext() {
return mapJoinContext;
}
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java Tue Apr 6 23:04:06 2010
@@ -43,7 +43,7 @@ public class QBParseInfo {
private final HashMap<String, ASTNode> aliasToSrc;
private final HashMap<String, ASTNode> nameToDest;
private final HashMap<String, TableSample> nameToSample;
- private final HashMap<String, ASTNode> destToSelExpr;
+ private final Map<String, ASTNode> destToSelExpr;
private final HashMap<String, ASTNode> destToWhereExpr;
private final HashMap<String, ASTNode> destToGroupby;
/**
@@ -83,7 +83,7 @@ public class QBParseInfo {
aliasToSrc = new HashMap<String, ASTNode>();
nameToDest = new HashMap<String, ASTNode>();
nameToSample = new HashMap<String, TableSample>();
- destToSelExpr = new HashMap<String, ASTNode>();
+ destToSelExpr = new LinkedHashMap<String, ASTNode>();
destToWhereExpr = new HashMap<String, ASTNode>();
destToGroupby = new HashMap<String, ASTNode>();
destToClusterby = new HashMap<String, ASTNode>();
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java Tue Apr 6 23:04:06 2010
@@ -50,11 +50,11 @@ import org.apache.hadoop.hive.ql.exec.Co
import org.apache.hadoop.hive.ql.exec.ConditionalTask;
import org.apache.hadoop.hive.ql.exec.ExecDriver;
import org.apache.hadoop.hive.ql.exec.FetchTask;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
import org.apache.hadoop.hive.ql.exec.FunctionInfo;
import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
import org.apache.hadoop.hive.ql.exec.GroupByOperator;
import org.apache.hadoop.hive.ql.exec.JoinOperator;
-import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
import org.apache.hadoop.hive.ql.exec.MapRedTask;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorFactory;
@@ -245,13 +245,15 @@ public class SemanticAnalyzer extends Ba
qb = pctx.getQB();
groupOpToInputTables = pctx.getGroupOpToInputTables();
prunedPartitions = pctx.getPrunedPartitions();
+ setLineageInfo(pctx.getLineageInfo());
}
public ParseContext getParseContext() {
return new ParseContext(conf, qb, ast, opToPartPruner, topOps, topSelOps,
- opParseCtx, joinContext, topToTable, loadTableWork, loadFileWork, ctx,
- idToTableNameMap, destTableId, uCtx, listMapJoinOpsNoReducer,
- groupOpToInputTables, prunedPartitions, opToSamplePruner);
+ opParseCtx, joinContext, topToTable, loadTableWork,
+ loadFileWork, ctx, idToTableNameMap, destTableId, uCtx,
+ listMapJoinOpsNoReducer, groupOpToInputTables, prunedPartitions,
+ opToSamplePruner);
}
@SuppressWarnings("nls")
@@ -3143,13 +3145,14 @@ public class SemanticAnalyzer extends Ba
QBMetaData qbm = qb.getMetaData();
Integer dest_type = qbm.getDestTypeForAlias(dest);
- Table dest_tab; // destination table if any
+ Table dest_tab = null; // destination table if any
String queryTmpdir = null; // the intermediate destination directory
Path dest_path = null; // the final destination directory
TableDesc table_desc = null;
int currentTableId = 0;
boolean isLocal = false;
SortBucketRSCtx rsCtx = new SortBucketRSCtx();
+ LoadTableDesc ltd = null;
switch (dest_type.intValue()) {
case QBMetaData.DEST_TABLE: {
@@ -3179,9 +3182,10 @@ public class SemanticAnalyzer extends Ba
// Create the work for moving the table
if (!isNonNativeTable) {
- loadTableWork.add(new LoadTableDesc(queryTmpdir, ctx
+ ltd = new LoadTableDesc(queryTmpdir, ctx
.getExternalTmpFileURI(dest_path.toUri()), table_desc,
- new HashMap<String, String>()));
+ new HashMap<String, String>());
+ loadTableWork.add(ltd);
}
if (!outputs.add(new WriteEntity(dest_tab))) {
throw new SemanticException(ErrorMsg.OUTPUT_SPECIFIED_MULTIPLE_TIMES
@@ -3205,9 +3209,10 @@ public class SemanticAnalyzer extends Ba
currentTableId = destTableId;
destTableId++;
- loadTableWork.add(new LoadTableDesc(queryTmpdir, ctx
+ ltd = new LoadTableDesc(queryTmpdir, ctx
.getExternalTmpFileURI(dest_path.toUri()), table_desc, dest_part
- .getSpec()));
+ .getSpec());
+ loadTableWork.add(ltd);
if (!outputs.add(new WriteEntity(dest_part))) {
throw new SemanticException(ErrorMsg.OUTPUT_SPECIFIED_MULTIPLE_TIMES
.getMsg(dest_tab.getTableName() + "@" + dest_part.getName()));
@@ -3355,6 +3360,12 @@ public class SemanticAnalyzer extends Ba
rsCtx.isMultiFileSpray(), rsCtx.getNumFiles(), rsCtx.getTotalFiles(), rsCtx.getPartnCols()),
fsRS, input), inputRR);
+
+ if (ltd != null && SessionState.get() != null) {
+ SessionState.get().getLineageState()
+ .mapDirToFop(ltd.getSourceDir(), (FileSinkOperator)output);
+ }
+
LOG.debug("Created FileSink Plan for clause: " + dest + "dest_path: "
+ dest_path + " row schema: " + inputRR.toString());
@@ -4833,8 +4844,7 @@ public class SemanticAnalyzer extends Ba
QBParseInfo qbp = qb.getParseInfo();
- TreeSet<String> ks = new TreeSet<String>();
- ks.addAll(qbp.getClauseNames());
+ TreeSet<String> ks = new TreeSet<String>(qbp.getClauseNames());
// For multi-group by with the same distinct, we ignore all user hints
// currently. It doesnt matter whether he has asked to do
@@ -5277,7 +5287,8 @@ public class SemanticAnalyzer extends Ba
colsEqual, alias, rwsch, qb.getMetaData(), null);
tableOp = OperatorFactory.getAndMakeChild(new FilterDesc(
samplePredicate, true, new sampleDesc(ts.getNumerator(), ts
- .getDenominator(), tabBucketCols, true)), top);
+ .getDenominator(), tabBucketCols, true)),
+ new RowSchema(rwsch.getColumnInfos()), top);
} else {
// need to add filter
// create tableOp to be filterDesc and set as child to 'top'
@@ -5285,7 +5296,8 @@ public class SemanticAnalyzer extends Ba
ExprNodeDesc samplePredicate = genSamplePredicate(ts, tabBucketCols,
colsEqual, alias, rwsch, qb.getMetaData(), null);
tableOp = OperatorFactory.getAndMakeChild(new FilterDesc(
- samplePredicate, true), top);
+ samplePredicate, true),
+ new RowSchema(rwsch.getColumnInfos()), top);
}
} else {
boolean testMode = conf.getBoolVar(HiveConf.ConfVars.HIVETESTMODE);
@@ -5316,7 +5328,8 @@ public class SemanticAnalyzer extends Ba
tableOp = OperatorFactory
.getAndMakeChild(new FilterDesc(samplePred, true,
new sampleDesc(tsSample.getNumerator(), tsSample
- .getDenominator(), tab.getBucketCols(), true)), top);
+ .getDenominator(), tab.getBucketCols(), true)),
+ new RowSchema(rwsch.getColumnInfos()), top);
LOG.info("No need for sample filter");
} else {
// The table is not bucketed, add a dummy filter :: rand()
@@ -5331,7 +5344,8 @@ public class SemanticAnalyzer extends Ba
ExprNodeDesc samplePred = genSamplePredicate(tsSample, null, false,
alias, rwsch, qb.getMetaData(), randFunc);
tableOp = OperatorFactory.getAndMakeChild(new FilterDesc(
- samplePred, true), top);
+ samplePred, true),
+ new RowSchema(rwsch.getColumnInfos()), top);
}
}
}
@@ -5616,8 +5630,9 @@ public class SemanticAnalyzer extends Ba
} else {
new ArrayList<MoveWork>();
for (LoadTableDesc ltd : loadTableWork) {
- mvTask.add(TaskFactory.get(new MoveWork(null, null, ltd, null, false),
- conf));
+ Task<MoveWork> tsk = TaskFactory.get(new MoveWork(null, null, ltd, null, false),
+ conf);
+ mvTask.add(tsk);
}
boolean oneLoadFile = true;
@@ -5950,8 +5965,8 @@ public class SemanticAnalyzer extends Ba
}
ParseContext pCtx = new ParseContext(conf, qb, child, opToPartPruner,
- topOps, topSelOps, opParseCtx, joinContext, topToTable, loadTableWork,
- loadFileWork, ctx, idToTableNameMap, destTableId, uCtx,
+ topOps, topSelOps, opParseCtx, joinContext, topToTable,
+ loadTableWork, loadFileWork, ctx, idToTableNameMap, destTableId, uCtx,
listMapJoinOpsNoReducer, groupOpToInputTables, prunedPartitions,
opToSamplePruner);
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MoveWork.java Tue Apr 6 23:04:06 2010
@@ -20,8 +20,9 @@ package org.apache.hadoop.hive.ql.plan;
import java.io.Serializable;
import java.util.HashSet;
-import java.util.Set;
+import java.util.List;
+import org.apache.hadoop.hive.metastore.api.Partition;
import org.apache.hadoop.hive.ql.hooks.ReadEntity;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
@@ -46,6 +47,11 @@ public class MoveWork implements Seriali
*/
protected HashSet<WriteEntity> outputs;
+ /**
+ * List of inserted partitions
+ */
+ protected List<Partition> movedParts;
+
public MoveWork() {
}
Added: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/LineageState.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/LineageState.java?rev=931363&view=auto
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/LineageState.java (added)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/LineageState.java Tue Apr 6 23:04:06 2010
@@ -0,0 +1,118 @@
+/**
+ * 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.session;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.DataContainer;
+import org.apache.hadoop.hive.ql.optimizer.lineage.LineageCtx.Index;
+
+/**
+ * LineageState. Contains all the information used to generate the
+ * lineage information for the post execution hooks.
+ *
+ */
+public class LineageState {
+
+ /**
+ * Mapping from the directory name to FileSinkOperator. This
+ * mapping is generated at the filesink operator creation
+ * time and is then later used to created the mapping from
+ * movetask to the set of filesink operators.
+ */
+ private final Map<String, FileSinkOperator> dirToFop;
+
+ /**
+ * The lineage context index for this query.
+ */
+ private Index index;
+
+ /**
+ * The lineage info structure that is used to pass the lineage
+ * information to the hooks.
+ */
+ private final LineageInfo linfo;
+
+ /**
+ * Constructor.
+ */
+ public LineageState() {
+ dirToFop = new HashMap<String, FileSinkOperator>();
+ linfo = new LineageInfo();
+ }
+
+ /**
+ * Adds a mapping from the load work to the file sink operator.
+ *
+ * @param dir The directory name.
+ * @param fop The file sink operator.
+ */
+ public void mapDirToFop(String dir, FileSinkOperator fop) {
+ dirToFop.put(dir, fop);
+ }
+
+ /**
+ * Set the lineage information for the associated directory.
+ *
+ * @param dir The directory containing the query results.
+ * @param dc The associated data container.
+ * @param cols The list of columns.
+ */
+ public void setLineage(String dir, DataContainer dc,
+ List<FieldSchema> cols) {
+ // First lookup the file sink operator from the load work.
+ FileSinkOperator fop = dirToFop.get(dir);
+
+ // Go over the associated fields and look up the dependencies
+ // by position in the row schema of the filesink operator.
+ if (fop == null) {
+ return;
+ }
+
+ List<ColumnInfo> signature = fop.getSchema().getSignature();
+ int i = 0;
+ for (FieldSchema fs : cols) {
+ linfo.putDependency(dc, fs, index.getDependency(fop, signature.get(i++)));
+ }
+ }
+
+ /**
+ * Gets the lineage information.
+ *
+ * @return LineageInfo.
+ */
+ public LineageInfo getLineageInfo() {
+ return linfo;
+ }
+
+ /**
+ * Sets the index for the lineage state.
+ *
+ * @param index The index derived from lineage context.
+ */
+ public void setIndex(Index index) {
+ this.index = index;
+ }
+}
Modified: hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java (original)
+++ hadoop/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java Tue Apr 6 23:04:06 2010
@@ -41,7 +41,7 @@ import org.apache.log4j.PropertyConfigur
/**
* SessionState encapsulates common data associated with a session.
- *
+ *
* Also provides support for a thread static session object that can be accessed
* from any point in the code to interact with the user and to retrieve
* configuration information
@@ -74,6 +74,20 @@ public class SessionState {
*/
private String commandType;
+ /**
+ * Lineage state.
+ */
+ LineageState ls;
+
+ /**
+ * Get the lineage state stored in this session.
+ *
+ * @return LineageState
+ */
+ public LineageState getLineageState() {
+ return ls;
+ }
+
public HiveConf getConf() {
return conf;
}
@@ -96,7 +110,7 @@ public class SessionState {
public SessionState(HiveConf conf) {
this.conf = conf;
-
+ ls = new LineageState();
}
public void setCmd(String cmdString) {
@@ -117,7 +131,7 @@ public class SessionState {
/**
* Singleton Session object per thread.
- *
+ *
**/
private static ThreadLocal<SessionState> tss = new ThreadLocal<SessionState>();
@@ -161,7 +175,7 @@ public class SessionState {
/**
* get hiveHitsory object which does structured logging.
- *
+ *
* @return The hive history object
*/
public HiveHistory getHiveHistory() {
@@ -196,11 +210,11 @@ public class SessionState {
* This class provides helper routines to emit informational and error
* messages to the user and log4j files while obeying the current session's
* verbosity levels.
- *
+ *
* NEVER write directly to the SessionStates standard output other than to
* emit result data DO use printInfo and printError provided by LogHelper to
* emit non result data strings.
- *
+ *
* It is perfectly acceptable to have global static LogHelper objects (for
* example - once per module) LogHelper always emits info/error to current
* session as required.
Modified: hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PostExecutePrinter.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PostExecutePrinter.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PostExecutePrinter.java (original)
+++ hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PostExecutePrinter.java Tue Apr 6 23:04:06 2010
@@ -18,8 +18,18 @@
package org.apache.hadoop.hive.ql.hooks;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.BaseColumnInfo;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.Dependency;
+import org.apache.hadoop.hive.ql.hooks.LineageInfo.DependencyKey;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
import org.apache.hadoop.security.UserGroupInformation;
@@ -30,9 +40,63 @@ import org.apache.hadoop.security.UserGr
*/
public class PostExecutePrinter implements PostExecute {
+ public class DependencyKeyComp implements
+ Comparator<Map.Entry<DependencyKey, Dependency>> {
+
+ @Override
+ public int compare(Map.Entry<DependencyKey, Dependency> o1,
+ Map.Entry<DependencyKey, Dependency> o2) {
+ if (o1 == null && o2 == null) {
+ return 0;
+ }
+ else if (o1 == null && o2 != null) {
+ return -1;
+ }
+ else if (o1 != null && o2 == null) {
+ return 1;
+ }
+ else {
+ // Both are non null.
+ // First compare the table names.
+ int ret = o1.getKey().getDataContainer().getTable().getTableName()
+ .compareTo(o2.getKey().getDataContainer().getTable().getTableName());
+
+ if (ret != 0) {
+ return ret;
+ }
+
+ // The table names match, so check on the partitions
+ if (!o1.getKey().getDataContainer().isPartition() &&
+ o2.getKey().getDataContainer().isPartition()) {
+ return -1;
+ }
+ else if (o1.getKey().getDataContainer().isPartition() &&
+ !o2.getKey().getDataContainer().isPartition()) {
+ return 1;
+ }
+
+ if (o1.getKey().getDataContainer().isPartition() &&
+ o2.getKey().getDataContainer().isPartition()) {
+ // Both are partitioned tables.
+ ret = o1.getKey().getDataContainer().getPartition().toString()
+ .compareTo(o2.getKey().getDataContainer().getPartition().toString());
+
+ if (ret != 0) {
+ return ret;
+ }
+ }
+
+ // The partitons are also the same so check the fieldschema
+ return (o1.getKey().getFieldSchema().getName().compareTo(
+ o2.getKey().getFieldSchema().getName()));
+ }
+ }
+ }
+
@Override
public void run(SessionState sess, Set<ReadEntity> inputs,
- Set<WriteEntity> outputs, UserGroupInformation ugi) throws Exception {
+ Set<WriteEntity> outputs, LineageInfo linfo,
+ UserGroupInformation ugi) throws Exception {
LogHelper console = SessionState.getConsole();
@@ -51,6 +115,50 @@ public class PostExecutePrinter implemen
for (WriteEntity we : outputs) {
console.printError("POSTHOOK: Output: " + we.toString());
}
+
+ // Also print out the generic lineage information if there is any
+ if (linfo != null) {
+ LinkedList<Map.Entry<DependencyKey, Dependency>> entry_list =
+ new LinkedList<Map.Entry<DependencyKey, Dependency>>(linfo.entrySet());
+ Collections.sort(entry_list, new DependencyKeyComp());
+ Iterator<Map.Entry<DependencyKey, Dependency>> iter = entry_list.iterator();
+ while(iter.hasNext()) {
+ Map.Entry<DependencyKey, Dependency> it = iter.next();
+ Dependency dep = it.getValue();
+ DependencyKey depK = it.getKey();
+
+ StringBuilder sb = new StringBuilder();
+ sb.append("POSTHOOK: Lineage: ");
+ if (depK.getDataContainer().isPartition()) {
+ Partition part = depK.getDataContainer().getPartition();
+ sb.append(part.getTableName());
+ sb.append(" PARTITION(");
+ int i = 0;
+ for (FieldSchema fs : depK.getDataContainer().getTable().getPartitionKeys()) {
+ if (i != 0) {
+ sb.append(",");
+ }
+ sb.append(fs.getName() + "=" + part.getValues().get(i++));
+ }
+ sb.append(")");
+ }
+ else {
+ sb.append(depK.getDataContainer().getTable().getTableName());
+ }
+ sb.append("." + depK.getFieldSchema().getName() + " " +
+ dep.getType() + " ");
+
+ sb.append("[");
+ for(BaseColumnInfo col: dep.getBaseCols()) {
+ sb.append("("+col.getTabAlias().getTable().getTableName() + ")"
+ + col.getTabAlias().getAlias() + "."
+ + col.getColumn() + ", ");
+ }
+ sb.append("]");
+
+ console.printError(sb.toString());
+ }
+ }
}
}
Modified: hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PreExecutePrinter.java
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PreExecutePrinter.java?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PreExecutePrinter.java (original)
+++ hadoop/hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/hooks/PreExecutePrinter.java Tue Apr 6 23:04:06 2010
@@ -32,7 +32,8 @@ public class PreExecutePrinter implement
@Override
public void run(SessionState sess, Set<ReadEntity> inputs,
- Set<WriteEntity> outputs, UserGroupInformation ugi) throws Exception {
+ Set<WriteEntity> outputs, UserGroupInformation ugi)
+ throws Exception {
LogHelper console = SessionState.getConsole();
Modified: hadoop/hive/trunk/ql/src/test/results/clientnegative/fileformat_void_input.q.out
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/results/clientnegative/fileformat_void_input.q.out?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/results/clientnegative/fileformat_void_input.q.out (original)
+++ hadoop/hive/trunk/ql/src/test/results/clientnegative/fileformat_void_input.q.out Tue Apr 6 23:04:06 2010
@@ -17,4 +17,6 @@ INSERT OVERWRITE TABLE dest1 SELECT src.
POSTHOOK: type: QUERY
POSTHOOK: Input: default@src
POSTHOOK: Output: default@dest1
+POSTHOOK: Lineage: dest1.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: dest1.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]
FAILED: Error in semantic analysis: line 3:20 Input Format must implement InputFormat dest1
Modified: hadoop/hive/trunk/ql/src/test/results/clientnegative/smb_bucketmapjoin.q.out
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/results/clientnegative/smb_bucketmapjoin.q.out?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/results/clientnegative/smb_bucketmapjoin.q.out (original)
+++ hadoop/hive/trunk/ql/src/test/results/clientnegative/smb_bucketmapjoin.q.out Tue Apr 6 23:04:06 2010
@@ -26,6 +26,8 @@ select * from src
POSTHOOK: type: QUERY
POSTHOOK: Input: default@src
POSTHOOK: Output: default@smb_bucket4_1
+POSTHOOK: Lineage: smb_bucket4_1.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: smb_bucket4_1.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]
PREHOOK: query: insert overwrite table smb_bucket4_2
select * from src
PREHOOK: type: QUERY
@@ -36,4 +38,8 @@ select * from src
POSTHOOK: type: QUERY
POSTHOOK: Input: default@src
POSTHOOK: Output: default@smb_bucket4_2
+POSTHOOK: Lineage: smb_bucket4_1.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: smb_bucket4_1.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]
+POSTHOOK: Lineage: smb_bucket4_2.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: smb_bucket4_2.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]
FAILED: Error in semantic analysis: Map Join cannot be performed with Outer join
Modified: hadoop/hive/trunk/ql/src/test/results/clientpositive/alter3.q.out
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/results/clientpositive/alter3.q.out?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/results/clientpositive/alter3.q.out (original)
+++ hadoop/hive/trunk/ql/src/test/results/clientpositive/alter3.q.out Tue Apr 6 23:04:06 2010
@@ -29,14 +29,16 @@ POSTHOOK: query: insert overwrite table
POSTHOOK: type: QUERY
POSTHOOK: Input: default@alter3_src
POSTHOOK: Output: default@alter3@pcol1=test_part/pcol2=test_part
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
PREHOOK: query: select * from alter3 where pcol1='test_part' and pcol2='test_part'
PREHOOK: type: QUERY
PREHOOK: Input: default@alter3@pcol1=test_part/pcol2=test_part
-PREHOOK: Output: file:/data/users/njain/hive5/hive5/build/ql/tmp/1483627653/10000
+PREHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-27_255_2797634165100554902/10000
POSTHOOK: query: select * from alter3 where pcol1='test_part' and pcol2='test_part'
POSTHOOK: type: QUERY
POSTHOOK: Input: default@alter3@pcol1=test_part/pcol2=test_part
-POSTHOOK: Output: file:/data/users/njain/hive5/hive5/build/ql/tmp/1483627653/10000
+POSTHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-27_255_2797634165100554902/10000
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
1 test_part test_part
2 test_part test_part
3 test_part test_part
@@ -49,32 +51,36 @@ POSTHOOK: query: alter table alter3 rena
POSTHOOK: type: ALTERTABLE_RENAME
POSTHOOK: Input: default@alter3
POSTHOOK: Output: default@alter3_renamed
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
PREHOOK: query: describe extended alter3_renamed
PREHOOK: type: DESCTABLE
POSTHOOK: query: describe extended alter3_renamed
POSTHOOK: type: DESCTABLE
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
col1 string
pcol1 string
pcol2 string
-Detailed Table Information Table(tableName:alter3_renamed, dbName:default, owner:njain, createTime:1253779705, lastAccessTime:0, retention:0, sd:StorageDescriptor(cols:[FieldSchema(name:col1, type:string, comment:null)], location:file:/data/users/njain/hive5/hive5/build/ql/test/data/warehouse/alter3_renamed, inputFormat:org.apache.hadoop.mapred.SequenceFileInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{serialization.format=1}), bucketCols:[], sortCols:[], parameters:{}), partitionKeys:[FieldSchema(name:pcol1, type:string, comment:null), FieldSchema(name:pcol2, type:string, comment:null)], parameters:{last_modified_by=njain,last_modified_time=1253779709})
+Detailed Table Information Table(tableName:alter3_renamed, dbName:default, owner:athusoo, createTime:1270515740, lastAccessTime:0, retention:0, sd:StorageDescriptor(cols:[FieldSchema(name:col1, type:string, comment:null)], location:file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/alter3_renamed, inputFormat:org.apache.hadoop.mapred.SequenceFileInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{serialization.format=1}), bucketCols:[], sortCols:[], parameters:{}), partitionKeys:[FieldSchema(name:pcol1, type:string, comment:null), FieldSchema(name:pcol2, type:string, comment:null)], parameters:{last_modified_by=athusoo,last_modified_time=1270515748,transient_lastDdlTime=1270515748}, viewOriginalText:null, viewExpandedText:null, tableType:MANAGED_TABLE)
PREHOOK: query: describe extended alter3_renamed partition (pCol1='test_part', pcol2='test_part')
PREHOOK: type: DESCTABLE
POSTHOOK: query: describe extended alter3_renamed partition (pCol1='test_part', pcol2='test_part')
POSTHOOK: type: DESCTABLE
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
col1 string
pcol1 string
pcol2 string
-Detailed Partition Information Partition(values:[test_part, test_part], dbName:default, tableName:alter3_renamed, createTime:0, lastAccessTime:0, sd:StorageDescriptor(cols:[FieldSchema(name:col1, type:string, comment:null)], location:file:/data/users/njain/hive5/hive5/build/ql/test/data/warehouse/alter3_renamed/pcol1=test_part/pcol2=test_part, inputFormat:org.apache.hadoop.mapred.SequenceFileInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{serialization.format=1}), bucketCols:[], sortCols:[], parameters:{}), parameters:{})
+Detailed Partition Information Partition(values:[test_part, test_part], dbName:default, tableName:alter3_renamed, createTime:1270515747, lastAccessTime:0, sd:StorageDescriptor(cols:[FieldSchema(name:col1, type:string, comment:null)], location:file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/alter3_renamed/pcol1=test_part/pcol2=test_part, inputFormat:org.apache.hadoop.mapred.SequenceFileInputFormat, outputFormat:org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat, compressed:false, numBuckets:-1, serdeInfo:SerDeInfo(name:null, serializationLib:org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, parameters:{serialization.format=1}), bucketCols:[], sortCols:[], parameters:{}), parameters:{transient_lastDdlTime=1270515747})
PREHOOK: query: select * from alter3_renamed where pcol1='test_part' and pcol2='test_part'
PREHOOK: type: QUERY
PREHOOK: Input: default@alter3_renamed@pcol1=test_part/pcol2=test_part
-PREHOOK: Output: file:/data/users/njain/hive5/hive5/build/ql/tmp/250127671/10000
+PREHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-28_356_3656682742332568668/10000
POSTHOOK: query: select * from alter3_renamed where pcol1='test_part' and pcol2='test_part'
POSTHOOK: type: QUERY
POSTHOOK: Input: default@alter3_renamed@pcol1=test_part/pcol2=test_part
-POSTHOOK: Output: file:/data/users/njain/hive5/hive5/build/ql/tmp/250127671/10000
+POSTHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-28_356_3656682742332568668/10000
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
1 test_part test_part
2 test_part test_part
3 test_part test_part
@@ -86,12 +92,15 @@ PREHOOK: type: DROPTABLE
POSTHOOK: query: drop table alter3_src
POSTHOOK: type: DROPTABLE
POSTHOOK: Output: default@alter3_src
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
PREHOOK: query: drop table alter3
PREHOOK: type: DROPTABLE
POSTHOOK: query: drop table alter3
POSTHOOK: type: DROPTABLE
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
PREHOOK: query: drop table alter3_renamed
PREHOOK: type: DROPTABLE
POSTHOOK: query: drop table alter3_renamed
POSTHOOK: type: DROPTABLE
POSTHOOK: Output: default@alter3_renamed
+POSTHOOK: Lineage: alter3 PARTITION(pcol1=test_part,pcol2=test_part).col1 SIMPLE [(alter3_src)alter3_src.FieldSchema(name:col1, type:string, comment:null), ]
Modified: hadoop/hive/trunk/ql/src/test/results/clientpositive/binary_output_format.q.out
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/results/clientpositive/binary_output_format.q.out?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/results/clientpositive/binary_output_format.q.out (original)
+++ hadoop/hive/trunk/ql/src/test/results/clientpositive/binary_output_format.q.out Tue Apr 6 23:04:06 2010
@@ -92,7 +92,7 @@ STAGE PLANS:
File Output Operator
compressed: false
GlobalTableId: 1
- directory: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10002
+ directory: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10002
NumFilesPerFileSink: 1
table:
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -103,22 +103,22 @@ STAGE PLANS:
columns.types string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveBinaryOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/dest1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/dest1
name dest1
serialization.ddl struct dest1 { string mydata}
serialization.format 1
serialization.last.column.takes.rest true
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449090
+ transient_lastDdlTime 1270515762
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: dest1
TotalFiles: 1
MultiFileSpray: false
Needs Tagging: false
Path -> Alias:
- file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src [src]
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/src [src]
Path -> Partition:
- file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/src
Partition
base file name: src
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -129,12 +129,12 @@ STAGE PLANS:
columns.types string:string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/src
name src
serialization.ddl struct src { string key, string value}
serialization.format 1
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449089
+ transient_lastDdlTime 1270515761
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -145,12 +145,12 @@ STAGE PLANS:
columns.types string:string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/src
name src
serialization.ddl struct src { string key, string value}
serialization.format 1
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449089
+ transient_lastDdlTime 1270515761
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: src
name: src
@@ -162,14 +162,14 @@ STAGE PLANS:
Move Operator
files:
hdfs directory: true
- source: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10002
- destination: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10000
+ source: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10002
+ destination: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10000
Stage: Stage-0
Move Operator
tables:
replace: true
- source: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10000
+ source: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10000
table:
input format: org.apache.hadoop.mapred.TextInputFormat
output format: org.apache.hadoop.hive.ql.io.HiveBinaryOutputFormat
@@ -179,21 +179,21 @@ STAGE PLANS:
columns.types string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveBinaryOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/dest1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/dest1
name dest1
serialization.ddl struct dest1 { string mydata}
serialization.format 1
serialization.last.column.takes.rest true
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449090
+ transient_lastDdlTime 1270515762
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: dest1
- tmp directory: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10001
+ tmp directory: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10001
Stage: Stage-2
Map Reduce
Alias -> Map Operator Tree:
- file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10002
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10002
Reduce Output Operator
sort order:
Map-reduce partition columns:
@@ -205,9 +205,9 @@ STAGE PLANS:
type: string
Needs Tagging: false
Path -> Alias:
- file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10002 [file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10002]
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10002 [file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10002]
Path -> Partition:
- file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10002
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10002
Partition
base file name: 10002
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -218,13 +218,13 @@ STAGE PLANS:
columns.types string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveBinaryOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/dest1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/dest1
name dest1
serialization.ddl struct dest1 { string mydata}
serialization.format 1
serialization.last.column.takes.rest true
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449090
+ transient_lastDdlTime 1270515762
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -235,13 +235,13 @@ STAGE PLANS:
columns.types string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveBinaryOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/dest1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/dest1
name dest1
serialization.ddl struct dest1 { string mydata}
serialization.format 1
serialization.last.column.takes.rest true
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449090
+ transient_lastDdlTime 1270515762
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: dest1
name: dest1
@@ -250,7 +250,7 @@ STAGE PLANS:
File Output Operator
compressed: false
GlobalTableId: 0
- directory: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-50_051_1929056605300008322/10000
+ directory: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-42_120_5258018612502461655/10000
NumFilesPerFileSink: 1
table:
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -261,13 +261,13 @@ STAGE PLANS:
columns.types string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveBinaryOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/dest1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/test/data/warehouse/dest1
name dest1
serialization.ddl struct dest1 { string mydata}
serialization.format 1
serialization.last.column.takes.rest true
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1266449090
+ transient_lastDdlTime 1270515762
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: dest1
TotalFiles: 1
@@ -302,16 +302,18 @@ FROM src
POSTHOOK: type: QUERY
POSTHOOK: Input: default@src
POSTHOOK: Output: default@dest1
+POSTHOOK: Lineage: dest1.mydata SCRIPT [(src)src.FieldSchema(name:key, type:string, comment:default), (src)src.FieldSchema(name:value, type:string, comment:default), ]
PREHOOK: query: -- Test the result
SELECT * FROM dest1
PREHOOK: type: QUERY
PREHOOK: Input: default@dest1
-PREHOOK: Output: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-54_510_6167292116608425875/10000
+PREHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-46_981_6800349673068207966/10000
POSTHOOK: query: -- Test the result
SELECT * FROM dest1
POSTHOOK: type: QUERY
POSTHOOK: Input: default@dest1
-POSTHOOK: Output: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-17_15-24-54_510_6167292116608425875/10000
+POSTHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_1/build/ql/scratchdir/hive_2010-04-05_18-02-46_981_6800349673068207966/10000
+POSTHOOK: Lineage: dest1.mydata SCRIPT [(src)src.FieldSchema(name:key, type:string, comment:default), (src)src.FieldSchema(name:value, type:string, comment:default), ]
238 val_238
86 val_86
311 val_311
@@ -817,3 +819,4 @@ PREHOOK: type: DROPTABLE
POSTHOOK: query: DROP TABLE dest1
POSTHOOK: type: DROPTABLE
POSTHOOK: Output: default@dest1
+POSTHOOK: Lineage: dest1.mydata SCRIPT [(src)src.FieldSchema(name:key, type:string, comment:default), (src)src.FieldSchema(name:value, type:string, comment:default), ]
Modified: hadoop/hive/trunk/ql/src/test/results/clientpositive/bucket1.q.out
URL: http://svn.apache.org/viewvc/hadoop/hive/trunk/ql/src/test/results/clientpositive/bucket1.q.out?rev=931363&r1=931362&r2=931363&view=diff
==============================================================================
--- hadoop/hive/trunk/ql/src/test/results/clientpositive/bucket1.q.out (original)
+++ hadoop/hive/trunk/ql/src/test/results/clientpositive/bucket1.q.out Tue Apr 6 23:04:06 2010
@@ -49,9 +49,9 @@ STAGE PLANS:
type: string
Needs Tagging: false
Path -> Alias:
- file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src [src]
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/src [src]
Path -> Partition:
- file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src
+ file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/src
Partition
base file name: src
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -62,12 +62,12 @@ STAGE PLANS:
columns.types string:string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/src
name src
serialization.ddl struct src { string key, string value}
serialization.format 1
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1267129667
+ transient_lastDdlTime 1270515753
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -78,12 +78,12 @@ STAGE PLANS:
columns.types string:string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/src
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/src
name src
serialization.ddl struct src { string key, string value}
serialization.format 1
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1267129667
+ transient_lastDdlTime 1270515753
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: src
name: src
@@ -99,7 +99,7 @@ STAGE PLANS:
File Output Operator
compressed: false
GlobalTableId: 1
- directory: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-25_12-27-48_091_9013372046421498540/10000
+ directory: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-34_525_873235169631577914/10000
NumFilesPerFileSink: 1
table:
input format: org.apache.hadoop.mapred.TextInputFormat
@@ -111,12 +111,12 @@ STAGE PLANS:
columns.types int:string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/bucket1_1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/bucket1_1
name bucket1_1
serialization.ddl struct bucket1_1 { i32 key, string value}
serialization.format 1
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1267129668
+ transient_lastDdlTime 1270515754
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: bucket1_1
TotalFiles: 1
@@ -126,7 +126,7 @@ STAGE PLANS:
Move Operator
tables:
replace: true
- source: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-25_12-27-48_091_9013372046421498540/10000
+ source: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-34_525_873235169631577914/10000
table:
input format: org.apache.hadoop.mapred.TextInputFormat
output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
@@ -137,15 +137,15 @@ STAGE PLANS:
columns.types int:string
file.inputformat org.apache.hadoop.mapred.TextInputFormat
file.outputformat org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat
- location file:/data/users/njain/hive1/hive1/build/ql/test/data/warehouse/bucket1_1
+ location file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/test/data/warehouse/bucket1_1
name bucket1_1
serialization.ddl struct bucket1_1 { i32 key, string value}
serialization.format 1
serialization.lib org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
- transient_lastDdlTime 1267129668
+ transient_lastDdlTime 1270515754
serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe
name: bucket1_1
- tmp directory: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-25_12-27-48_091_9013372046421498540/10001
+ tmp directory: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-34_525_873235169631577914/10001
PREHOOK: query: insert overwrite table bucket1_1
@@ -158,14 +158,18 @@ select * from src
POSTHOOK: type: QUERY
POSTHOOK: Input: default@src
POSTHOOK: Output: default@bucket1_1
+POSTHOOK: Lineage: bucket1_1.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: bucket1_1.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]
PREHOOK: query: select * from bucket1_1 order by key
PREHOOK: type: QUERY
PREHOOK: Input: default@bucket1_1
-PREHOOK: Output: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-25_12-27-52_282_2132999711625321779/10000
+PREHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-39_954_6102377035167270592/10000
POSTHOOK: query: select * from bucket1_1 order by key
POSTHOOK: type: QUERY
POSTHOOK: Input: default@bucket1_1
-POSTHOOK: Output: file:/data/users/njain/hive1/hive1/build/ql/scratchdir/hive_2010-02-25_12-27-52_282_2132999711625321779/10000
+POSTHOOK: Output: file:/data/users/athusoo/apache_workspaces/hive_trunk_ws1/.ptest_0/build/ql/scratchdir/hive_2010-04-05_18-02-39_954_6102377035167270592/10000
+POSTHOOK: Lineage: bucket1_1.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: bucket1_1.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]
0 val_0
0 val_0
0 val_0
@@ -671,3 +675,5 @@ PREHOOK: type: DROPTABLE
POSTHOOK: query: drop table bucket1_1
POSTHOOK: type: DROPTABLE
POSTHOOK: Output: default@bucket1_1
+POSTHOOK: Lineage: bucket1_1.key EXPRESSION [(src)src.FieldSchema(name:key, type:string, comment:default), ]
+POSTHOOK: Lineage: bucket1_1.value SIMPLE [(src)src.FieldSchema(name:value, type:string, comment:default), ]