You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ju...@apache.org on 2012/06/19 02:33:12 UTC
svn commit: r1351547 [1/2] - in /pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/rela...
Author: julien
Date: Tue Jun 19 00:33:11 2012
New Revision: 1351547
URL: http://svn.apache.org/viewvc?rev=1351547&view=rev
Log:
PIG-2699: Reduce the number of instances of Load and Store Funcs down to 2+1 (julien)
Added:
pig/trunk/test/org/apache/pig/TestLoadStoreFuncLifeCycle.java
pig/trunk/test/org/apache/pig/newplan/logical/relational/LOTestHelper.java
Removed:
pig/trunk/src/org/apache/pig/newplan/logical/rules/LoadStoreFuncDupSignatureValidator.java
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/PigServer.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java
pig/trunk/src/org/apache/pig/newplan/logical/relational/LOLoad.java
pig/trunk/src/org/apache/pig/newplan/logical/relational/LOStore.java
pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
pig/trunk/src/org/apache/pig/newplan/logical/rules/PartitionFilterOptimizer.java
pig/trunk/src/org/apache/pig/newplan/logical/rules/TypeCastInserter.java
pig/trunk/src/org/apache/pig/newplan/logical/visitor/ScalarVisitor.java
pig/trunk/src/org/apache/pig/parser/FunctionType.java
pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java
pig/trunk/src/org/apache/pig/parser/QueryParserUtils.java
pig/trunk/test/org/apache/pig/parser/TestScalarVisitor.java
pig/trunk/test/org/apache/pig/test/TestInputOutputFileValidator.java
pig/trunk/test/org/apache/pig/test/TestLoad.java
pig/trunk/test/org/apache/pig/test/TestLogToPhyCompiler.java
pig/trunk/test/org/apache/pig/test/TestLogicalPlanBuilder.java
pig/trunk/test/org/apache/pig/test/TestMRCompiler.java
pig/trunk/test/org/apache/pig/test/TestNewPlanFilterRule.java
pig/trunk/test/org/apache/pig/test/TestNewPlanListener.java
pig/trunk/test/org/apache/pig/test/TestNewPlanLogicalOptimizer.java
pig/trunk/test/org/apache/pig/test/TestNewPlanOperatorPlan.java
pig/trunk/test/org/apache/pig/test/TestTypeCheckingValidatorNewLP.java
pig/trunk/test/org/apache/pig/test/Util.java
pig/trunk/test/org/apache/pig/test/utils/TypeCheckingTestUtil.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Tue Jun 19 00:33:11 2012
@@ -24,6 +24,8 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-2699: Reduce the number of instances of Load and Store Funcs down to 2+1 (julien)
+
PIG-2166: UDFs to join a bag (hluu via daijy)
PIG-2651: Provide a much easier to use accumulator interface (jcoveney via daijy)
Modified: pig/trunk/src/org/apache/pig/PigServer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/PigServer.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/PigServer.java (original)
+++ pig/trunk/src/org/apache/pig/PigServer.java Tue Jun 19 00:33:11 2012
@@ -943,7 +943,7 @@ public class PigServer {
currDAG.buildPlan( alias );
try {
- QueryParserUtils.attachStorePlan( currDAG.lp, filename, func, currDAG.getOperator( alias ), alias, pigContext );
+ QueryParserUtils.attachStorePlan(scope, currDAG.lp, filename, func, currDAG.getOperator( alias ), alias, pigContext);
currDAG.compile();
return executeCompiledLogicalPlan();
} catch (PigException e) {
@@ -1225,7 +1225,7 @@ public class PigServer {
if( !isBatchOn() || alias != null ) {
// MRCompiler needs a store to be the leaf - hence
// add a store to the plan to explain
- QueryParserUtils.attachStorePlan( currDAG.lp, "fakefile", null, currDAG.getOperator( alias ),
+ QueryParserUtils.attachStorePlan(scope, currDAG.lp, "fakefile", null, currDAG.getOperator( alias ),
"fake", pigContext );
}
currDAG.compile();
@@ -1628,9 +1628,9 @@ public class PigServer {
}
private void compile(LogicalPlan lp) throws FrontendException {
- new ColumnAliasConversionVisitor( lp ).visit();
- new SchemaAliasVisitor( lp ).visit();
- new ScalarVisitor( lp, pigContext ).visit();
+ new ColumnAliasConversionVisitor(lp).visit();
+ new SchemaAliasVisitor(lp).visit();
+ new ScalarVisitor(lp, pigContext, scope).visit();
// TODO: move optimizer here from HExecuteEngine.
// TODO: input/output validation visitor
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Tue Jun 19 00:33:11 2012
@@ -56,7 +56,6 @@ import org.apache.pig.newplan.logical.re
import org.apache.pig.newplan.logical.relational.LogicalPlan;
import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator;
import org.apache.pig.newplan.logical.rules.InputOutputFileValidator;
-import org.apache.pig.newplan.logical.rules.LoadStoreFuncDupSignatureValidator;
import org.apache.pig.newplan.logical.visitor.SortInfoSetter;
import org.apache.pig.newplan.logical.visitor.StoreAliasSetter;
import org.apache.pig.pen.POOptimizeDisabler;
@@ -269,11 +268,7 @@ public class HExecutionEngine {
optimizerRules.add("AddForEach");
optimizerRules.add("GroupByConstParallelSetter");
}
-
- // Check if we have duplicate signature
- LoadStoreFuncDupSignatureValidator loadStoreFuncDupSignatureValidator = new LoadStoreFuncDupSignatureValidator(plan);
- loadStoreFuncDupSignatureValidator.validate();
-
+
StoreAliasSetter storeAliasSetter = new StoreAliasSetter( plan );
storeAliasSetter.visit();
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Tue Jun 19 00:33:11 2012
@@ -252,7 +252,7 @@ public class MapReduceLauncher extends L
// Set the thread UDFContext so registered classes are available.
final UDFContext udfContext = UDFContext.getUDFContext();
- Thread jcThread = new Thread(jc) {
+ Thread jcThread = new Thread(jc, "JobControl") {
@Override
public void run() {
UDFContext.setUdfContext(udfContext.clone()); //PIG-2576
@@ -650,7 +650,7 @@ public class MapReduceLauncher extends L
} catch (Exception e) {
String errMsg = "Could not resolve error that occured when launching map reduce job: "
+ jobControlExceptionStackTrace;
- jobControlException = new RuntimeException(errMsg);
+ jobControlException = new RuntimeException(errMsg, throwable);
}
}
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POStore.java Tue Jun 19 00:33:11 2012
@@ -293,4 +293,8 @@ public class POStore extends PhysicalOpe
public boolean disableCounter() {
return disableCounter;
}
+
+ public void setStoreFunc(StoreFuncInterface storeFunc) {
+ this.storer = storeFunc;
+ }
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/relational/LOLoad.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOLoad.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LOLoad.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LOLoad.java Tue Jun 19 00:33:11 2012
@@ -40,33 +40,58 @@ import org.apache.pig.newplan.logical.Ut
public class LOLoad extends LogicalRelationalOperator {
private LogicalSchema scriptSchema;
- private FileSpec fs;
+ private final FileSpec fs;
private transient LoadFunc loadFunc;
transient private Configuration conf;
- private LogicalSchema determinedSchema;
+ private final LogicalSchema determinedSchema;
private List<Integer> requiredFields = null;
private boolean castInserted = false;
private LogicalSchema uidOnlySchema;
- private String schemaFile = null;
- private String signature = null;
+ private final String schemaFile;
+ private final String signature;
private long limit = -1;
/**
+ * used for pattern matching
*
- * @param loader FuncSpec for load function to use for this load.
* @param schema schema user specified in script, or null if not
* specified.
* @param plan logical plan this load is part of.
*/
- public LOLoad(FileSpec loader, LogicalSchema schema, LogicalPlan plan, Configuration conf) {
- super("LOLoad", plan);
- scriptSchema = schema;
- fs = loader;
- if (loader != null)
- schemaFile = loader.getFileName();
- this.conf = conf;
+ public LOLoad(LogicalSchema schema, LogicalPlan plan) {
+ this(null, schema, plan, null, null, null);
}
-
+
+ /**
+ * Used from the LogicalPlanBuilder
+ *
+ * @param loader FuncSpec for load function to use for this load.
+ * @param schema schema user specified in script, or null if not specified.
+ * @param plan logical plan this load is part of.
+ * @param conf
+ * @param loadFunc the LoadFunc that was instantiated from loader
+ * @param signature the signature that will be passed to the LoadFunc
+ */
+ public LOLoad(FileSpec loader, LogicalSchema schema, LogicalPlan plan, Configuration conf, LoadFunc loadFunc, String signature) {
+ super("LOLoad", plan);
+ this.scriptSchema = schema;
+ this.fs = loader;
+ this.schemaFile = loader == null ? null : loader.getFileName();
+ this.conf = conf;
+ this.loadFunc = loadFunc;
+ this.signature = signature;
+ if (loadFunc != null) {
+ this.loadFunc.setUDFContextSignature(signature);
+ try {
+ this.determinedSchema = getSchemaFromMetaData();
+ } catch (FrontendException e) {
+ throw new RuntimeException("Can not retrieve schema from loader " + loadFunc, e);
+ }
+ } else {
+ this.determinedSchema = null;
+ }
+ }
+
public String getSchemaFile() {
return schemaFile;
}
@@ -105,10 +130,6 @@ public class LOLoad extends LogicalRelat
return schema;
LogicalSchema originalSchema = null;
-
- if (determinedSchema==null) {
- determinedSchema = getSchemaFromMetaData();
- }
if (scriptSchema != null && determinedSchema != null) {
originalSchema = LogicalSchema.merge(scriptSchema, determinedSchema, LogicalSchema.MergeMode.LoadForEach);
@@ -158,15 +179,13 @@ public class LOLoad extends LogicalRelat
return null;
}
- @Override
- public void setAlias(String alias) {
- super.setAlias(alias);
-
- // set the schema in this method using the new alias assigned
- storeScriptSchema();
- if (signature==null)
- signature = alias;
- }
+ @Override
+ public void setAlias(String alias) {
+ super.setAlias(alias);
+
+ // set the schema in this method using the new alias assigned
+ storeScriptSchema();
+ }
/**
* This method will store the scriptSchema:Schema using ObjectSerializer to
@@ -270,16 +289,6 @@ public class LOLoad extends LogicalRelat
return signature;
}
- /***
- * This method is called by Pig logical planner to setup UDFContext signature.
- * So that loadFunc can use signature to store its own configurations in UDFContext.
- * This is not intend to be called by users
- */
- public void setSignature(String signature) {
- this.signature = signature;
- loadFunc.setUDFContextSignature(signature);
- }
-
public LogicalSchema getScriptSchema() {
return scriptSchema;
}
@@ -291,4 +300,5 @@ public class LOLoad extends LogicalRelat
public void setLimit(long limit) {
this.limit = limit;
}
+
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/relational/LOStore.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOStore.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LOStore.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LOStore.java Tue Jun 19 00:33:11 2012
@@ -17,42 +17,31 @@
*/
package org.apache.pig.newplan.logical.relational;
-//import org.apache.commons.logging.Log;
-//import org.apache.commons.logging.LogFactory;
-import org.apache.pig.FuncSpec;
import org.apache.pig.SortInfo;
import org.apache.pig.StoreFuncInterface;
-import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileSpec;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.PlanVisitor;
public class LOStore extends LogicalRelationalOperator {
- private static final long serialVersionUID = 2L;
- private FileSpec output;
-
- // If we know how to reload the store, here's how. The lFile
+ private final FileSpec output;
+
+ // If we know how to reload the store, here's how. The lFile
// FileSpec is set in PigServer.postProcess. It can be used to
// reload this store, if the optimizer has the need.
private FileSpec mInputSpec;
- private String signature;
+ private final String signature;
private boolean isTmpStore;
private SortInfo sortInfo;
- transient private StoreFuncInterface storeFunc;
-
- //private static Log log = LogFactory.getLog(LOStore.class);
-
- public LOStore(LogicalPlan plan) {
- super("LOStore", plan);
- }
-
- public LOStore(LogicalPlan plan, FileSpec outputFileSpec) {
- super("LOStore", plan);
+ private final StoreFuncInterface storeFunc;
- output = outputFileSpec;
- storeFunc = (StoreFuncInterface) PigContext.instantiateFuncFromSpec(outputFileSpec.getFuncSpec());
+ public LOStore(LogicalPlan plan, FileSpec outputFileSpec, StoreFuncInterface storeFunc, String signature) {
+ super("LOStore", plan);
+ this.output = outputFileSpec;
+ this.storeFunc = storeFunc;
+ this.signature = signature;
}
public FileSpec getOutputSpec() {
@@ -123,23 +112,9 @@ public class LOStore extends LogicalRela
public String getSignature() {
return signature;
}
-
- public void setSignature(String sig) {
- signature = sig;
- storeFunc.setStoreFuncUDFContextSignature(signature);
- }
- public static String constructSignature(String alias, String filename, FuncSpec funcSpec) {
- return alias+"_"+filename+"_"+funcSpec.toString();
+ public FileSpec getFileSpec() {
+ return output;
}
- public FileSpec getFileSpec() {
- return output;
- }
-
- @Override
- public void setAlias(String alias) {
- this.alias = alias;
- setSignature(constructSignature(alias, output.getFileName(), output.getFuncSpec()));
- }
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java Tue Jun 19 00:33:11 2012
@@ -677,6 +677,7 @@ public class LogToPhyTranslationVisitor
store.setSignature(loStore.getSignature());
store.setSortInfo(loStore.getSortInfo());
store.setIsTmpStore(loStore.isTmpStore());
+ store.setStoreFunc(loStore.getStoreFunc());
store.setSchema(Util.translateSchema( loStore.getSchema() ));
Modified: pig/trunk/src/org/apache/pig/newplan/logical/rules/PartitionFilterOptimizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/rules/PartitionFilterOptimizer.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/rules/PartitionFilterOptimizer.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/rules/PartitionFilterOptimizer.java Tue Jun 19 00:33:11 2012
@@ -85,7 +85,7 @@ public class PartitionFilterOptimizer ex
protected OperatorPlan buildPattern() {
// match each foreach.
LogicalPlan plan = new LogicalPlan();
- LogicalRelationalOperator load = new LOLoad (null, null, plan, null );
+ LogicalRelationalOperator load = new LOLoad (null, plan);
plan.add( load );
// LogicalRelationalOperator filter = new LOFilter( plan );
// plan.add( filter );
Modified: pig/trunk/src/org/apache/pig/newplan/logical/rules/TypeCastInserter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/rules/TypeCastInserter.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/rules/TypeCastInserter.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/rules/TypeCastInserter.java Tue Jun 19 00:33:11 2012
@@ -55,7 +55,7 @@ public abstract class TypeCastInserter e
LogicalPlan plan = new LogicalPlan();
LogicalRelationalOperator op;
if (getOperatorClassName().equals(LOLoad.class.getName()))
- op = new LOLoad(null, null, plan, null);
+ op = new LOLoad(null, plan);
else // LOStream
op = new LOStream(plan, null, null, null);
plan.add(op);
Modified: pig/trunk/src/org/apache/pig/newplan/logical/visitor/ScalarVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/ScalarVisitor.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/visitor/ScalarVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/ScalarVisitor.java Tue Jun 19 00:33:11 2012
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.util.List;
import org.apache.pig.FuncSpec;
+import org.apache.pig.StoreFuncInterface;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.io.FileSpec;
@@ -38,6 +39,7 @@ import org.apache.pig.newplan.logical.ex
import org.apache.pig.newplan.logical.optimizer.AllExpressionVisitor;
import org.apache.pig.newplan.logical.relational.LOStore;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
+import org.apache.pig.parser.LogicalPlanBuilder;
/**
* Logical plan visitor which handles scalar projections. It will find or create a LOStore
@@ -46,10 +48,12 @@ import org.apache.pig.newplan.logical.re
*/
public class ScalarVisitor extends AllExpressionVisitor {
private final PigContext pigContext;
+ private final String scope;
- public ScalarVisitor(OperatorPlan plan, PigContext pigContext) throws FrontendException {
+ public ScalarVisitor(OperatorPlan plan, PigContext pigContext, String scope) throws FrontendException {
super( plan, new DependencyOrderWalker( plan ) );
this.pigContext = pigContext;
+ this.scope = scope;
}
@Override
@@ -84,7 +88,10 @@ public class ScalarVisitor extends AllEx
} catch (IOException e) {
throw new PlanValidationException( expr, "Failed to process scalar" + e);
}
- store = new LOStore( lp, fileSpec );
+ StoreFuncInterface stoFunc = (StoreFuncInterface)PigContext.instantiateFuncFromSpec(funcSpec);
+ String sig = LogicalPlanBuilder.newOperatorKey(scope);
+ stoFunc.setStoreFuncUDFContextSignature(sig);
+ store = new LOStore(lp, fileSpec, stoFunc, sig);
store.setTmpStore(true);
lp.add( store );
lp.connect( refOp, store );
Modified: pig/trunk/src/org/apache/pig/parser/FunctionType.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/FunctionType.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/FunctionType.java (original)
+++ pig/trunk/src/org/apache/pig/parser/FunctionType.java Tue Jun 19 00:33:11 2012
@@ -33,28 +33,32 @@ class FunctionType {
public static final byte PIGTOSTREAMFUNC = 32;
public static final byte STREAMTOPIGFUNC = 64;
- public static void tryCasting(Object func, byte funcType) throws Exception {
+ public static void tryCasting(Class<?> func, byte funcType) {
+ Class<?> typeClass;
switch(funcType) {
case FunctionType.EVALFUNC:
- EvalFunc evalFunc = (EvalFunc) func;
+ typeClass = EvalFunc.class;
break;
case FunctionType.COMPARISONFUNC:
- ComparisonFunc comparisonFunc = (ComparisonFunc) func;
+ typeClass = ComparisonFunc.class;
break;
case FunctionType.LOADFUNC:
- LoadFunc loadFunc = (LoadFunc) func;
+ typeClass = LoadFunc.class;
break;
case FunctionType.STOREFUNC:
- StoreFuncInterface storeFunc = (StoreFuncInterface) func;
+ typeClass = StoreFuncInterface.class;
break;
case FunctionType.PIGTOSTREAMFUNC:
- PigToStream ptsFunc = (PigToStream) func;
+ typeClass = PigToStream.class;
break;
case FunctionType.STREAMTOPIGFUNC:
- StreamToPig stpFunc = (StreamToPig) func;
+ typeClass = StreamToPig.class;
break;
default:
- throw new Exception("Received an unknown function type: " + funcType);
+ throw new IllegalArgumentException("Received an unknown function type: " + funcType);
+ }
+ if (!typeClass.isAssignableFrom(func)) {
+ throw new ClassCastException(func + " does not implement " + typeClass);
}
}
Modified: pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java (original)
+++ pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java Tue Jun 19 00:33:11 2012
@@ -106,6 +106,10 @@ public class LogicalPlanBuilder {
private static NodeIdGenerator nodeIdGen = NodeIdGenerator.getGenerator();
+ public static long getNextId(String scope) {
+ return nodeIdGen.getNextNodeId( scope );
+ }
+
LogicalPlanBuilder(PigContext pigContext, String scope, Map<String, String> fileNameMap,
IntStream input) {
this.pigContext = pigContext;
@@ -635,44 +639,46 @@ public class LogicalPlanBuilder {
op.setInnerFlags( flags );
alias = buildOp( loc, op, alias, inputAliases, partitioner );
expandAndResetVisitor(loc, op);
+
return alias;
}
-
- private String getAbolutePathForLoad(String filename, FuncSpec funcSpec)
- throws IOException, URISyntaxException {
- if( funcSpec == null ){
- funcSpec = new FuncSpec( PigStorage.class.getName() );
- }
-
- LoadFunc loFunc = (LoadFunc)PigContext.instantiateFuncFromSpec( funcSpec );
- String sig = QueryParserUtils.constructFileNameSignature( filename, funcSpec );
- String absolutePath = fileNameMap.get( sig );
- if (absolutePath == null) {
- absolutePath = loFunc.relativeToAbsolutePath( filename, QueryParserUtils.getCurrentDir( pigContext ) );
- if (absolutePath!=null) {
- QueryParserUtils.setHdfsServers( absolutePath, pigContext );
- }
- fileNameMap.put( sig, absolutePath );
- }
-
- return absolutePath;
- }
-
String buildLoadOp(SourceLocation loc, String alias, String filename, FuncSpec funcSpec, LogicalSchema schema)
throws ParserValidationException {
- String absolutePath = filename;
+ String absolutePath;
+ LoadFunc loFunc;
try {
- absolutePath = getAbolutePathForLoad( filename, funcSpec );
+ FuncSpec instantiatedFuncSpec =
+ funcSpec == null ?
+ new FuncSpec(PigStorage.class.getName()) :
+ funcSpec;
+
+ loFunc = (LoadFunc)PigContext.instantiateFuncFromSpec(instantiatedFuncSpec);
+ String sig = QueryParserUtils.constructFileNameSignature(filename, instantiatedFuncSpec);
+ absolutePath = fileNameMap.get(sig);
+ if (absolutePath == null) {
+ absolutePath = loFunc.relativeToAbsolutePath( filename, QueryParserUtils.getCurrentDir( pigContext ) );
+
+ if (absolutePath!=null) {
+ QueryParserUtils.setHdfsServers( absolutePath, pigContext );
+ }
+ fileNameMap.put( sig, absolutePath );
+ }
} catch(Exception ex) {
throw new ParserValidationException( intStream, loc, ex );
}
-
- FileSpec loader = new FileSpec( absolutePath, funcSpec );
- LOLoad op = new LOLoad( loader, schema, plan, ConfigurationUtil.toConfiguration( pigContext.getProperties() ) );
+
+ FileSpec loader = new FileSpec(absolutePath, funcSpec);
+ LOLoad op = new LOLoad(
+ loader,
+ schema,
+ plan,
+ ConfigurationUtil.toConfiguration(pigContext.getProperties()),
+ loFunc,
+ alias + "_" + newOperatorKey());
return buildOp( loc, op, alias, new ArrayList<String>(), null );
}
-
+
private String buildOp(SourceLocation loc, LogicalRelationalOperator op, String alias,
String inputAlias, String partitioner) throws ParserValidationException {
List<String> inputAliases = new ArrayList<String>();
@@ -699,41 +705,41 @@ public class LogicalPlanBuilder {
return op.getAlias();
}
- private String getAbolutePathForStore(String inputAlias, String filename, FuncSpec funcSpec)
- throws IOException, URISyntaxException {
- if( funcSpec == null ){
- funcSpec = new FuncSpec( PigStorage.class.getName() );
- }
-
- Object obj = PigContext.instantiateFuncFromSpec( funcSpec );
- StoreFuncInterface stoFunc = (StoreFuncInterface)obj;
- String sig = QueryParserUtils.constructSignature( inputAlias, filename, funcSpec);
- stoFunc.setStoreFuncUDFContextSignature( sig );
- String absolutePath = fileNameMap.get( sig );
- if (absolutePath == null) {
- absolutePath = stoFunc.relToAbsPathForStoreLocation( filename,
- QueryParserUtils.getCurrentDir( pigContext ) );
- if (absolutePath!=null) {
- QueryParserUtils.setHdfsServers( absolutePath, pigContext );
- }
- fileNameMap.put( sig, absolutePath );
- }
-
- return absolutePath;
- }
-
String buildStoreOp(SourceLocation loc, String alias, String inputAlias, String filename, FuncSpec funcSpec)
throws ParserValidationException {
- String absPath = filename;
try {
- absPath = getAbolutePathForStore( inputAlias, filename, funcSpec );
+ FuncSpec instantiatedFuncSpec =
+ funcSpec == null ?
+ new FuncSpec(PigStorage.class.getName()):
+ funcSpec;
+
+ StoreFuncInterface stoFunc = (StoreFuncInterface)PigContext.instantiateFuncFromSpec(instantiatedFuncSpec);
+ String sig = inputAlias + "_" + newOperatorKey();
+ stoFunc.setStoreFuncUDFContextSignature(sig);
+ String absolutePath = fileNameMap.get(sig);
+ if (absolutePath == null) {
+ absolutePath = stoFunc.relToAbsPathForStoreLocation(
+ filename,
+ QueryParserUtils.getCurrentDir(pigContext));
+ if (absolutePath!=null) {
+ QueryParserUtils.setHdfsServers(absolutePath, pigContext);
+ }
+ fileNameMap.put(sig, absolutePath);
+ }
+ FileSpec fileSpec = new FileSpec(absolutePath, funcSpec);
+ LOStore op = new LOStore(plan, fileSpec, stoFunc, sig);
+ return buildOp(loc, op, alias, inputAlias, null);
} catch(Exception ex) {
- throw new ParserValidationException( intStream, loc, ex );
+ throw new ParserValidationException(intStream, loc, ex);
}
-
- FileSpec fileSpec = new FileSpec( absPath, funcSpec );
- LOStore op = new LOStore( plan, fileSpec );
- return buildOp( loc, op, alias, inputAlias, null );
+ }
+
+ private String newOperatorKey() {
+ return new OperatorKey( scope, getNextId() ).toString();
+ }
+
+ public static String newOperatorKey(String scope) {
+ return new OperatorKey( scope, getNextId(scope)).toString();
}
LOForEach createForeachOp() {
@@ -983,7 +989,7 @@ public class LogicalPlanBuilder {
void setAlias(LogicalRelationalOperator op, String alias) {
if( alias == null )
- alias = new OperatorKey( scope, getNextId() ).toString();
+ alias = newOperatorKey();
op.setAlias( alias );
}
@@ -1006,17 +1012,17 @@ public class LogicalPlanBuilder {
}
private void validateFuncSpec(SourceLocation loc, FuncSpec funcSpec, byte ft) throws RecognitionException {
- switch( ft ) {
+ switch (ft) {
case FunctionType.COMPARISONFUNC:
case FunctionType.LOADFUNC:
case FunctionType.STOREFUNC:
case FunctionType.STREAMTOPIGFUNC:
case FunctionType.PIGTOSTREAMFUNC:
- Object func = PigContext.instantiateFuncFromSpec( funcSpec );
try{
- FunctionType.tryCasting( func, ft );
+ Class<?> func = PigContext.resolveClassName(funcSpec.getClassName());
+ FunctionType.tryCasting(func, ft);
} catch(Exception ex){
- throw new ParserValidationException( intStream, loc, ex );
+ throw new ParserValidationException(intStream, loc, ex);
}
}
}
@@ -1196,32 +1202,33 @@ public class LogicalPlanBuilder {
LogicalExpression buildUDF(SourceLocation loc, LogicalExpressionPlan plan,
String funcName, List<LogicalExpression> args)
throws RecognitionException {
- Object func;
+
+ Class<?> func;
try {
- func = pigContext.instantiateFuncFromAlias( funcName );
- FunctionType.tryCasting( func, FunctionType.EVALFUNC );
+ func = pigContext.getClassForAlias(funcName);
+ FunctionType.tryCasting(func, FunctionType.EVALFUNC);
} catch (Exception e) {
- throw new PlanGenerationFailureException( intStream, loc, e );
+ throw new PlanGenerationFailureException(intStream, loc, e);
}
- FuncSpec funcSpec = pigContext.getFuncSpecFromAlias( funcName );
+ FuncSpec funcSpec = pigContext.getFuncSpecFromAlias(funcName);
LogicalExpression le;
if( funcSpec == null ) {
- funcName = func.getClass().getName();
- funcSpec = new FuncSpec( funcName );
+ funcName = func.getName();
+ funcSpec = new FuncSpec(funcName);
//this point is only reached if there was no DEFINE statement for funcName
//in which case, we pass that information along
- le = new UserFuncExpression( plan, funcSpec, args, false );
+ le = new UserFuncExpression(plan, funcSpec, args, false);
} else {
- le = new UserFuncExpression( plan, funcSpec, args, true );
+ le = new UserFuncExpression(plan, funcSpec, args, true);
}
- le.setLocation( loc );
+ le.setLocation(loc);
return le;
}
private long getNextId() {
- return nodeIdGen.getNextNodeId( scope );
+ return getNextId(scope);
}
static LOFilter createNestedFilterOp(LogicalPlan plan) {
Modified: pig/trunk/src/org/apache/pig/parser/QueryParserUtils.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/QueryParserUtils.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/QueryParserUtils.java (original)
+++ pig/trunk/src/org/apache/pig/parser/QueryParserUtils.java Tue Jun 19 00:33:11 2012
@@ -61,7 +61,7 @@ public class QueryParserUtils {
return str;
}
- public static void attachStorePlan(LogicalPlan lp, String fileName, String func,
+ public static void attachStorePlan(String scope, LogicalPlan lp, String fileName, String func,
Operator input, String alias, PigContext pigContext) throws FrontendException {
if( func == null ) {
func = PigStorage.class.getName();
@@ -69,12 +69,13 @@ public class QueryParserUtils {
FuncSpec funcSpec = new FuncSpec( func );
StoreFuncInterface stoFunc = (StoreFuncInterface)PigContext.instantiateFuncFromSpec( funcSpec );
- stoFunc.setStoreFuncUDFContextSignature( LOStore.constructSignature( alias, fileName, funcSpec ) );
-
+
fileName = removeQuotes( fileName );
FileSpec fileSpec = new FileSpec( fileName, funcSpec );
- LOStore store = new LOStore( lp, fileSpec );
- store.setAlias( alias );
+ String sig = alias + "_" + LogicalPlanBuilder.newOperatorKey(scope);
+ stoFunc.setStoreFuncUDFContextSignature(sig);
+ LOStore store = new LOStore(lp, fileSpec, stoFunc, sig);
+ store.setAlias(alias);
try {
stoFunc.relToAbsPathForStoreLocation( fileName, getCurrentDir( pigContext ) );
@@ -169,14 +170,10 @@ public class QueryParserUtils {
return fileName + "_" + funcSpec.toString();
}
- static String constructSignature(String alias, String filename, FuncSpec funcSpec) {
- return alias + "_" + filename + "_" + funcSpec.toString();
- }
-
static String generateErrorHeader(RecognitionException ex, String filename) {
return new SourceLocation( filename, ex.line, ex.charPositionInLine ).toString();
}
-
+
@SuppressWarnings({ "unchecked", "rawtypes" })
static void replaceNodeWithNodeList(Tree oldNode, CommonTree newTree,
String fileName) {
Added: pig/trunk/test/org/apache/pig/TestLoadStoreFuncLifeCycle.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/TestLoadStoreFuncLifeCycle.java?rev=1351547&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/TestLoadStoreFuncLifeCycle.java (added)
+++ pig/trunk/test/org/apache/pig/TestLoadStoreFuncLifeCycle.java Tue Jun 19 00:33:11 2012
@@ -0,0 +1,369 @@
+/**
+ * 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.pig;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertTrue;
+import static org.apache.pig.builtin.mock.Storage.tuple;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
+import org.apache.pig.builtin.mock.Storage;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.junit.Test;
+
+public class TestLoadStoreFuncLifeCycle {
+
+ public static String loaderSignature;
+ public static String storerSignature;
+
+ abstract public static class InstrumentedStorage extends Storage implements LoadPushDown {
+
+ protected final int id;
+
+ abstract void incCall();
+
+ private void logCaller(Object... params) {
+ incCall();
+ TestLoadStoreFuncLifeCycle.logCaller(id, this.getClass(), params);
+ }
+
+ public InstrumentedStorage(int id) {
+ super();
+ this.id = id;
+ logCaller();
+ }
+
+ @Override
+ public String relativeToAbsolutePath(String location, Path curDir)
+ throws IOException {
+ logCaller(location, curDir);
+ return super.relativeToAbsolutePath(location, curDir);
+ }
+
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ logCaller(location, job);
+ super.setLocation(location, job);
+ }
+
+ @Override
+ public InputFormat<?, ?> getInputFormat() throws IOException {
+ logCaller();
+ return super.getInputFormat();
+ }
+
+ @Override
+ public LoadCaster getLoadCaster() throws IOException {
+ logCaller();
+ return super.getLoadCaster();
+ }
+
+ @Override
+ public void prepareToRead(@SuppressWarnings("rawtypes") RecordReader reader, PigSplit split)
+ throws IOException {
+ logCaller(reader, split);
+ super.prepareToRead(reader, split);
+ }
+
+ @Override
+ public Tuple getNext() throws IOException {
+ logCaller();
+ return super.getNext();
+ }
+
+ @Override
+ public void setUDFContextSignature(String signature) {
+ logCaller(signature);
+ if (TestLoadStoreFuncLifeCycle.loaderSignature == null) {
+ TestLoadStoreFuncLifeCycle.loaderSignature = signature;
+ } else {
+ assertEquals(TestLoadStoreFuncLifeCycle.loaderSignature, signature);
+ }
+ super.setUDFContextSignature(signature);
+ }
+
+ @Override
+ public ResourceSchema getSchema(String location, Job job)
+ throws IOException {
+ logCaller(location, job);
+ return super.getSchema(location, job);
+ }
+
+ @Override
+ public ResourceStatistics getStatistics(String location, Job job)
+ throws IOException {
+ logCaller(location, job);
+ return super.getStatistics(location, job);
+ }
+
+ @Override
+ public String[] getPartitionKeys(String location, Job job)
+ throws IOException {
+ logCaller(location, job);
+ return super.getPartitionKeys(location, job);
+ }
+
+ @Override
+ public void setPartitionFilter(Expression partitionFilter)
+ throws IOException {
+ logCaller(partitionFilter);
+ super.setPartitionFilter(partitionFilter);
+ }
+
+ @Override
+ public List<OperatorSet> getFeatures() {
+ logCaller();
+ return null;
+ }
+
+ @Override
+ public RequiredFieldResponse pushProjection(
+ RequiredFieldList requiredFieldList) throws FrontendException {
+ logCaller(requiredFieldList);
+ return null;
+ }
+
+ @Override
+ public String relToAbsPathForStoreLocation(String location, Path curDir)
+ throws IOException {
+ logCaller(location, curDir);
+ return super.relToAbsPathForStoreLocation(location, curDir);
+ }
+
+ @Override
+ public OutputFormat<?, ?> getOutputFormat() throws IOException {
+ logCaller();
+ return super.getOutputFormat();
+ }
+
+ @Override
+ public void setStoreLocation(String location, Job job) throws IOException {
+ logCaller(location, job);
+ super.setStoreLocation(location, job);
+ }
+
+ @Override
+ public void checkSchema(ResourceSchema s) throws IOException {
+ logCaller(s);
+ super.checkSchema(s);
+ }
+
+ @Override
+ public void prepareToWrite(@SuppressWarnings("rawtypes") RecordWriter writer) throws IOException {
+ logCaller(writer);
+ super.prepareToWrite(writer);
+ }
+
+ @Override
+ public void putNext(Tuple t) throws IOException {
+ logCaller(t);
+ super.putNext(t);
+ }
+
+ @Override
+ public void setStoreFuncUDFContextSignature(String signature) {
+ logCaller(signature);
+ if (TestLoadStoreFuncLifeCycle.storerSignature == null) {
+ TestLoadStoreFuncLifeCycle.storerSignature = signature;
+ } else {
+ assertEquals(TestLoadStoreFuncLifeCycle.storerSignature, signature);
+ }
+ super.setStoreFuncUDFContextSignature(signature);
+ }
+
+ @Override
+ public void cleanupOnFailure(String location, Job job) throws IOException {
+ logCaller(location, job);
+ super.cleanupOnFailure(location, job);
+ }
+
+ @Override
+ public void storeStatistics(ResourceStatistics stats, String location,
+ Job job) throws IOException {
+ logCaller(stats, location, job);
+ super.storeStatistics(stats, location, job);
+ }
+
+ @Override
+ public void storeSchema(ResourceSchema schema, String location, Job job)
+ throws IOException {
+ logCaller(schema, location, job);
+ super.storeSchema(schema, location, job);
+ }
+
+ }
+
+ public static class Loader extends InstrumentedStorage {
+ static int count;
+ static int callCount;
+
+ public Loader() {
+ super(++ count);
+ }
+
+ @Override
+ void incCall() {
+ ++callCount;
+ }
+
+ }
+
+ public static class Storer extends InstrumentedStorage {
+ static int count;
+ static int callCount;
+
+ public Storer() {
+ super(++ count);
+ }
+
+ @Override
+ void incCall() {
+ ++callCount;
+ }
+
+ }
+
+ private static final int MAX_PARAM_SIZE = 70;
+
+ static List<String> calls = new ArrayList<String>();
+ static List<String> constructorCallers = new ArrayList<String>();
+
+ private static void logCaller(int id, Class<?> clazz, Object[] params) {
+ StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
+ int i = 0;
+ // skip the stackframes that where after the call to logCaller (included)
+ while (!stackTrace[i].toString().startsWith(InstrumentedStorage.class.getName() + ".logCaller")) {
+ ++i;
+ }
+ StackTraceElement called = stackTrace[i + 1];
+ String calledClass = clazz.getSimpleName();
+
+ String paramsString = null;
+ for (Object param : params) {
+ String paramString = String.valueOf(param);
+ if (paramString.length() > MAX_PARAM_SIZE || paramString.contains("\n")) {
+ int end = paramString.indexOf('\n');
+ if (end == -1 || end > MAX_PARAM_SIZE) {
+ end = MAX_PARAM_SIZE;
+ }
+ paramString = paramString.substring(0, end) + "...";
+ }
+ if (paramsString == null) {
+ paramsString = "(";
+ } else {
+ paramsString += ", ";
+ }
+ paramsString += paramString;
+ }
+ if (paramsString == null) {
+ paramsString = "()";
+ } else {
+ paramsString += ")";
+ }
+ String call = calledClass + "[" + id + "]." + called.getMethodName();
+ calls.add(call + paramsString /*+ " called by " + findSalient(stackTrace)*/);
+ if (called.getMethodName().equals("<init>")) {
+ constructorCallers.add(call + " called by " + findSalient(stackTrace));
+ }
+ }
+
+ /**
+ * helper method for debugging
+ * @param stackTrace the stack trace of the constructor call
+ * @return the first interesting stack frame of where the constructor was called from
+ */
+ private static String findSalient(StackTraceElement[] stackTrace) {
+ String message = "";
+ int count = 0;
+ // when finding the caller we want to keep only the pig calls to shorten the report
+ for (StackTraceElement el : stackTrace) {
+ String cl = el.getClassName();
+ if (cl.startsWith("org.apache.pig")
+ && !cl.startsWith(TestLoadStoreFuncLifeCycle.class.getName())
+ ) {
+ message += "\n" + el.toString();
+ ++ count;
+ }
+ if (count > 15) {
+ break;
+ }
+ }
+
+ return message;
+ }
+
+ @Test
+ public void testLoadStoreFunc() throws Exception {
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ Data data = Storage.resetData(pigServer.getPigContext());
+ data.set("foo",
+ tuple("a"),
+ tuple("b"),
+ tuple("c")
+ );
+
+ pigServer.registerQuery(
+ "A = LOAD 'foo' USING " + TestLoadStoreFuncLifeCycle.class.getName() + "$Loader();\n" +
+ "STORE A INTO 'bar' USING " + TestLoadStoreFuncLifeCycle.class.getName() + "$Storer();");
+
+ List<Tuple> out = data.get("bar");
+
+ assertEquals("a", out.get(0).get(0));
+ assertEquals("b", out.get(1).get(0));
+ assertEquals("c", out.get(2).get(0));
+
+ assertTrue("loader instanciation count increasing: " + Loader.count, Loader.count <= 3);
+ assertTrue("storer instanciation count increasing: " + Storer.count, Storer.count <= 3);
+
+ }
+
+ /**
+ * prints out a report on instantiation sites and method calls
+ */
+ public static void main(String[] args) throws Exception {
+ new TestLoadStoreFuncLifeCycle().testLoadStoreFunc();
+ System.out.println("report:");
+ System.out.println(Loader.count + " instances of Loader");
+ System.out.println(Loader.callCount + " calls to Loader");
+ System.out.println(Storer.count + " instances of Storer");
+ System.out.println(Storer.callCount + " calls to Storer");
+ System.out.println();
+ System.out.println("all calls:");
+ for (String caller : calls) {
+ System.out.println(caller);
+ }
+ System.out.println();
+ System.out.println("constructor calls:");
+ for (String caller : constructorCallers) {
+ System.out.println(caller);
+ }
+ }
+
+}
Added: pig/trunk/test/org/apache/pig/newplan/logical/relational/LOTestHelper.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/newplan/logical/relational/LOTestHelper.java?rev=1351547&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/newplan/logical/relational/LOTestHelper.java (added)
+++ pig/trunk/test/org/apache/pig/newplan/logical/relational/LOTestHelper.java Tue Jun 19 00:33:11 2012
@@ -0,0 +1,12 @@
+package org.apache.pig.newplan.logical.relational;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileSpec;
+
+public class LOTestHelper {
+ public static LOLoad newLOLoad(FileSpec loader, LogicalSchema schema, LogicalPlan plan, Configuration conf) {
+ return new LOLoad(loader, schema, plan, conf, (LoadFunc)PigContext.instantiateFuncFromSpec(loader.getFuncSpec()), null);
+ }
+}
Modified: pig/trunk/test/org/apache/pig/parser/TestScalarVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/parser/TestScalarVisitor.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/parser/TestScalarVisitor.java (original)
+++ pig/trunk/test/org/apache/pig/parser/TestScalarVisitor.java Tue Jun 19 00:33:11 2012
@@ -88,7 +88,7 @@ public class TestScalarVisitor {
LogicalPlan plan = ParserTestingUtils.generateLogicalPlan( query );
PigContext pc = new PigContext( ExecType.LOCAL, new Properties() );
pc.connect();
- ScalarVisitor visitor = new ScalarVisitor( plan, pc );
+ ScalarVisitor visitor = new ScalarVisitor(plan, pc, "test");
visitor.visit();
return plan;
}
Modified: pig/trunk/test/org/apache/pig/test/TestInputOutputFileValidator.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestInputOutputFileValidator.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestInputOutputFileValidator.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestInputOutputFileValidator.java Tue Jun 19 00:33:11 2012
@@ -17,6 +17,8 @@
*/
package org.apache.pig.test;
+import static org.apache.pig.newplan.logical.relational.LOTestHelper.newLOLoad;
+
import java.io.* ;
import java.util.Iterator;
import java.util.Properties;
@@ -26,6 +28,7 @@ import org.apache.pig.FuncSpec;
import org.apache.pig.PigException;
import org.apache.pig.PigServer;
import org.apache.pig.ResourceSchema;
+import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.datastorage.ElementDescriptor;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
@@ -243,9 +246,9 @@ public class TestInputOutputFileValidato
new FileSpec(inputFile, new FuncSpec("org.apache.pig.builtin.PigStorage")) ;
FileSpec filespec2 =
new FileSpec(outputFile, new FuncSpec("org.apache.pig.builtin.PigStorage"));
- LOLoad load = new LOLoad( filespec1, null, plan,
+ LOLoad load = newLOLoad( filespec1, null, plan,
ConfigurationUtil.toConfiguration(dfs.getConfiguration())) ;
- LOStore store = new LOStore(plan, filespec2) ;
+ LOStore store = new LOStore(plan, filespec2, (StoreFuncInterface)PigContext.instantiateFuncFromSpec(filespec2.getFuncSpec()), null) ;
plan.add(load) ;
plan.add(store) ;
Modified: pig/trunk/test/org/apache/pig/test/TestLoad.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestLoad.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestLoad.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestLoad.java Tue Jun 19 00:33:11 2012
@@ -164,7 +164,7 @@ public class TestLoad extends junit.fram
@Test
public void testLoadRemoteAbsAuth() throws Exception {
pc = servers[0].getPigContext();
- checkLoadPath("hdfs://localhost:9000/test","/test");
+ checkLoadPath(cluster.getFileSystem().getUri()+"/test","/test");
}
@Test
Modified: pig/trunk/test/org/apache/pig/test/TestLogToPhyCompiler.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestLogToPhyCompiler.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestLogToPhyCompiler.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestLogToPhyCompiler.java Tue Jun 19 00:33:11 2012
@@ -20,13 +20,16 @@ package org.apache.pig.test;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
+import java.io.FileNotFoundException;
import java.io.FileOutputStream;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.Random;
+import org.apache.commons.lang.CharSet;
import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.SortColInfo;
@@ -45,15 +48,9 @@ import org.apache.pig.newplan.logical.re
import org.junit.Test;
/**
- * All new tests should be included at the end of the existing test cases. This is to ensure that
- * nodeIdGenerator produces the same sequence as in the golden files for the previously existing test cases
*
- * To generate golden files, use the following code :
- * ByteArrayOutputStream baos = new ByteArrayOutputStream();
- * pp.explain(baos);
- * FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Union.gld");
- * fos.write(baos.toByteArray());
- *
+ * To generate golden files, update generate to true
+ *
*/
public class TestLogToPhyCompiler extends junit.framework.TestCase {
File A;
@@ -65,6 +62,7 @@ public class TestLogToPhyCompiler extend
private boolean generate = false;
PigServer pigServer = null;
+ private static final int MAX_SIZE = 100000;;
@Override
@@ -73,335 +71,163 @@ public class TestLogToPhyCompiler extend
pc.connect();
}
- @Test // Commented out due to PIG-2020
- public void testComplexForeach() throws Exception {
- String query = "C = foreach (load 'a' as (a:bag{} ) ) {" +
- "B = FILTER $0 BY ($1 == $2);" +
- "generate B;" +
- "};" + "store C into 'output';";
- LogicalPlan plan = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(plan);
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
+ private String compilePlan(String query, String goldenFile, boolean checkOneLeaf)
+ throws Exception, FrontendException, FileNotFoundException,
+ IOException {
+ LogicalPlan lp = buildPlan(query);
+ PhysicalPlan pp = buildPhysicalPlan(lp);
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/ComplexForeach.gld");
- fos.write(baos.toByteArray());
- return;
+ if (checkOneLeaf) {
+ //Ensure that there is only 1 leaf node
+ assertEquals(1, pp.getLeaves().size());
}
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/ComplexForeach.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testComplexForeach");
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
- }
-
- public void testSort() throws Exception {
- String query = "store (order (load 'a') by $0) into 'output';";
- LogicalPlan plan = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(plan);
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
pp.explain(baos);
baos.write((int)'\n');
String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
+ compiledPlan = removedUnaffectingChanges(compiledPlan);
+
+ generateGolden(goldenFile, compiledPlan);
+ return compiledPlan;
+ }
+ private void generateGolden(String goldenFile, String data)
+ throws FileNotFoundException, IOException {
if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Sort.gld");
- fos.write(baos.toByteArray());
- return;
+ FileOutputStream fos = new FileOutputStream(goldenFile);
+ fos.write(data.getBytes("UTF-8"));
+ fos.close();
}
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Sort.gld");
+ }
+
+ private String readGolden(int MAX_SIZE, String goldenFile)
+ throws FileNotFoundException, IOException {
+ FileInputStream fis = new FileInputStream(goldenFile);
byte[] b = new byte[MAX_SIZE];
int len = fis.read(b);
+ fis.close();
String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
+ goldenPlan = removedUnaffectingChanges(goldenPlan);
+ return goldenPlan;
+ }
+ private String removedUnaffectingChanges(String compiledPlan) {
+ return compiledPlan
+ .replaceAll("test-[0-9]*: ","test-: ")
+ .replaceAll("Load(.*)","Load()")
+ .replaceAll("Store(.*)","Store()")
+ .replaceAll("scope-[0-9]*\\n", "\n");
+ }
+
+ private void checkAgainstGolden(String query, String goldenFile,
+ boolean checkOneLeaf, String testName) throws Exception, FrontendException,
+ FileNotFoundException, IOException {
+ goldenFile = "test/org/apache/pig/test/data/GoldenFiles/" + goldenFile;
+ String compiledPlan = compilePlan(query, goldenFile, checkOneLeaf);
+ String goldenPlan = readGolden(MAX_SIZE, goldenFile);
System.out.println();
System.out.println(compiledPlan);
- System.out.println("-------------testSort");
-
+ System.out.println("-------------" + testName);
assertEquals(compiledPlan, goldenPlan);
}
+
+ @Test // Commented out due to PIG-2020
+ public void testComplexForeach() throws Exception {
+ checkAgainstGolden(
+ "C = foreach (load 'a' as (a:bag{} ) ) {" +
+ "B = FILTER $0 BY ($1 == $2);" +
+ "generate B;" +
+ "};" +
+ "store C into 'output';",
+ "ComplexForeach.gld",
+ false,
+ "testComplexForeach");
+ }
+
+ public void testSort() throws Exception {
+ checkAgainstGolden(
+ "store (order (load 'a') by $0) into 'output';",
+ "Sort.gld",
+ false,
+ "testSort");
+ }
+
public void testDistinct() throws Exception {
- String query = "store( distinct (load 'a') ) into 'output';";
- LogicalPlan plan = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(plan);
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Distinct.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Distinct.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testDistinct");
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "store( distinct (load 'a') ) into 'output';",
+ "Distinct.gld",
+ false,
+ "testDistinct");
}
public void testCogroup() throws Exception {
- System.out.println("testCogroup");
- String query = "A = cogroup (load 'a') by ($0 + $1, $0 - $1), (load 'b') by ($0 + $1, $0 - $1);"
- + "store A into 'output';";
- LogicalPlan plan = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(plan);
-
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Cogroup.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Cogroup.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------");
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "A = cogroup (load 'a') by ($0 + $1, $0 - $1), (load 'b') by ($0 + $1, $0 - $1);" +
+ "store A into 'output';",
+ "Cogroup.gld",
+ false,
+ "testCogroup");
}
public void testArithmetic() throws Exception {
-
- String query = "A = foreach (load 'A') generate $0 + $1 + 5, $0 - 5 - $1, 'hello';" +
- "store A into 'output';";
- LogicalPlan lp = buildPlan(query);
-
- PhysicalPlan pp = buildPhysicalPlan(lp);
-
- //Ensure that there is only 1 leaf node
- assertEquals(1, pp.getLeaves().size());
-
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Arithmetic.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Arithmetic.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------");
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "A = foreach (load 'A') generate $0 + $1 + 5, $0 - 5 - $1, 'hello';" +
+ "store A into 'output';",
+ "Arithmetic.gld",
+ true,
+ "testArithmetic");
}
-
- public void testComparison() throws Exception {
- String query = "A = filter (load 'a' using " + PigStorage.class.getName() + "(':')) by $0 + $1 > ($0 - $1) * (4 / 2);" +
- "store A into 'output';";
- LogicalPlan lp = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(lp);
-
- //Ensure that there is only 1 leaf node
- assertEquals(1, pp.getLeaves().size());
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Comparison.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Comparison.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testComparison");
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
+ public void testComparison() throws Exception {
+ checkAgainstGolden(
+ "A = filter (load 'a' using " + PigStorage.class.getName() + "(':')) by $0 + $1 > ($0 - $1) * (4 / 2);" +
+ "store A into 'output';",
+ "Comparison.gld",
+ true,
+ "testComparison");
}
@Test
public void testBinCond() throws Exception {
- String query = "A = foreach (load 'a') generate ($1 == '3'? $2 + $3 : $2 - $3) ;" +
- "store A into 'output';";
- LogicalPlan lp = buildPlan(query);
-
- PhysicalPlan pp = buildPhysicalPlan(lp);
-
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/BinCond.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/BinCond.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testBinCond");
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "A = foreach (load 'a') generate ($1 == '3'? $2 + $3 : $2 - $3) ;" +
+ "store A into 'output';",
+ "BinCond.gld",
+ false,
+ "testBinCond");
}
-
-
+
@Test
public void testGenerate() throws Exception {
- String query = "A = foreach (load 'a') generate ($1+$2), ($1-$2), ($1*$2), ($1/$2), ((int)$1%(int)$2), -($1) ;" +
- "store A into 'output';";
- LogicalPlan lp = buildPlan(query);
-
- PhysicalPlan pp = buildPhysicalPlan(lp);
-
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Generate.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Generate.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testGenerate");
-
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "A = foreach (load 'a') generate ($1+$2), ($1-$2), ($1*$2), ($1/$2), ((int)$1%(int)$2), -($1) ;" +
+ "store A into 'output';",
+ "Generate.gld",
+ false,
+ "testGenerate");
}
@Test
public void testUnion() throws Exception {
- String query = "A = union (load 'a'), (load 'b'), (load 'c');" +
- "store A into 'output';";
- LogicalPlan lp = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(lp);
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Union.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Union.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testUnion");
-
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "A = union (load 'a'), (load 'b'), (load 'c');" +
+ "store A into 'output';",
+ "Union.gld",
+ false,
+ "testUnion");
}
@Test
public void testSplit() throws Exception {
String query = "split (load 'a') into x if $0 < '7', y if $0 > '7';" +
"store x into 'output';";
- LogicalPlan plan = buildPlan(query);
-
- PhysicalPlan pp = buildPhysicalPlan(plan);
+ String goldenFile = "test/org/apache/pig/test/data/GoldenFiles/Split1.gld";
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Split1.gld");
- fos.write(baos.toByteArray());
- return;
- }
+ String compiledPlan = compilePlan(query, goldenFile, false);
- FileInputStream fis1 = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Split1.gld");
+ FileInputStream fis1 = new FileInputStream(goldenFile);
FileInputStream fis2 = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Split2.gld");
byte[] b1 = new byte[MAX_SIZE];
byte[] b2 = new byte[MAX_SIZE];
@@ -410,8 +236,8 @@ public class TestLogToPhyCompiler extend
//System.out.println("Length of first plan = " + len + " of second = " + test);
String goldenPlan1 = new String(b1, 0, len);
String goldenPlan2 = new String(b2, 0, len);
- goldenPlan1 = goldenPlan1.replaceAll("Load(.*)","Load()");
- goldenPlan2 = goldenPlan2.replaceAll("Load(.*)","Load()");
+ goldenPlan1 = removedUnaffectingChanges(goldenPlan1);
+ goldenPlan2 = removedUnaffectingChanges(goldenPlan2);
System.out.println();
System.out.println(compiledPlan);
@@ -444,17 +270,15 @@ public class TestLogToPhyCompiler extend
PhysicalPlan pp = buildPhysicalPlan(plan);
- int MAX_SIZE = 100000;
ByteArrayOutputStream baos = new ByteArrayOutputStream();
pp.explain(baos);
baos.write((int)'\n');
String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
+ compiledPlan = removedUnaffectingChanges(compiledPlan);
if(generate){
FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/IsNull1.gld");
fos.write(baos.toByteArray());
- return;
}
FileInputStream fis1 = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/IsNull1.gld");
@@ -466,8 +290,8 @@ public class TestLogToPhyCompiler extend
//System.out.println("Length of first plan = " + len + " of second = " + test + " Length of compiled plan = " + compiledPlan.length());
String goldenPlan1 = new String(b1, 0, len);
String goldenPlan2 = new String(b2, 0, len);
- goldenPlan1 = goldenPlan1.replaceAll("Load(.*)","Load()");
- goldenPlan2 = goldenPlan2.replaceAll("Load(.*)","Load()");
+ goldenPlan1 = removedUnaffectingChanges(goldenPlan1);
+ goldenPlan2 = removedUnaffectingChanges(goldenPlan2);
System.out.println();
System.out.println(compiledPlan);
@@ -491,35 +315,11 @@ public class TestLogToPhyCompiler extend
@Test
public void testLimit() throws Exception {
- String query = "store( limit (load 'a') 5 ) into 'output';";
- LogicalPlan plan = buildPlan(query);
- PhysicalPlan pp = buildPhysicalPlan(plan);
-
- int MAX_SIZE = 100000;
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- pp.explain(baos);
- baos.write((int)'\n');
- String compiledPlan = baos.toString();
- compiledPlan = compiledPlan.replaceAll("Load(.*)","Load()").replaceAll("Store(.*)","Store()");
-
- if(generate){
- FileOutputStream fos = new FileOutputStream("test/org/apache/pig/test/data/GoldenFiles/Limit.gld");
- fos.write(baos.toByteArray());
- return;
- }
-
- FileInputStream fis = new FileInputStream("test/org/apache/pig/test/data/GoldenFiles/Limit.gld");
- byte[] b = new byte[MAX_SIZE];
- int len = fis.read(b);
- String goldenPlan = new String(b, 0, len);
- goldenPlan = goldenPlan.replaceAll("Load(.*)","Load()");
-
- System.out.println();
- System.out.println(compiledPlan);
- System.out.println("-------------testLimit");
-
- //System.out.println(compiledPlan.compareTo(goldenPlan)==0);
- assertEquals(compiledPlan, goldenPlan);
+ checkAgainstGolden(
+ "store( limit (load 'a') 5 ) into 'output';",
+ "Limit.gld",
+ false,
+ "testLimit");
}
/**
Modified: pig/trunk/test/org/apache/pig/test/TestLogicalPlanBuilder.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestLogicalPlanBuilder.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestLogicalPlanBuilder.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestLogicalPlanBuilder.java Tue Jun 19 00:33:11 2012
@@ -2084,14 +2084,17 @@ public class TestLogicalPlanBuilder {
LogicalPlan plan = buildPlan( query );
Operator op = plan.getSinks().get(0);
LOLoad load = (LOLoad)plan.getPredecessors(op).get(0);
- Assert.assertTrue(((PigStorageWithSchema)(load).getLoadFunc()).getUDFContextSignature().equals("a"));
+ // the signature is now a unique string of the format "{alias}_{scope id}-{id}" example: "a_12-0"
+ String udfContextSignature = ((PigStorageWithSchema)(load).getLoadFunc()).getUDFContextSignature();
+ Assert.assertTrue(udfContextSignature, udfContextSignature.matches("a_[0-9]*-[0-9]*"));
query = " b = load '1.txt' using org.apache.pig.test.PigStorageWithSchema();" +
"store b into 'output';";
plan = buildPlan(query);
op = plan.getSinks().get(0);
load = (LOLoad)plan.getPredecessors(op).get(0);
- Assert.assertTrue(((PigStorageWithSchema)(load).getLoadFunc()).getUDFContextSignature().equals("b"));
+ udfContextSignature = ((PigStorageWithSchema)(load).getLoadFunc()).getUDFContextSignature();
+ Assert.assertTrue(udfContextSignature, udfContextSignature.matches("b_[0-9]*-[0-9]*"));
}
@Test
Modified: pig/trunk/test/org/apache/pig/test/TestMRCompiler.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestMRCompiler.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestMRCompiler.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestMRCompiler.java Tue Jun 19 00:33:11 2012
@@ -1077,7 +1077,17 @@ public class TestMRCompiler extends juni
System.out.println("Golden");
System.out.println("<<<" + goldenPlan + ">>>");
System.out.println("-------------");
- assertEquals(goldenPlan, compiledPlan);
+ assertEquals(removeSignature(goldenPlan), removeSignature(compiledPlan));
+ }
+
+ /**
+ * this removes the signature from the serialized plan
+ * changing the way the unique signature is generated should not break this test
+ * @param plan the plan to canonicalize
+ * @return the cleaned up plan
+ */
+ private String removeSignature(String plan) {
+ return plan.replaceAll("','','[^']*','scope','true'\\)\\)", "','','','scope','true'))");
}
public static class TestCollectableLoadFunc extends PigStorage implements CollectableLoadFunc {
Modified: pig/trunk/test/org/apache/pig/test/TestNewPlanFilterRule.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestNewPlanFilterRule.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestNewPlanFilterRule.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestNewPlanFilterRule.java Tue Jun 19 00:33:11 2012
@@ -80,7 +80,7 @@ public class TestNewPlanFilterRule {
schema.getField(0).uid = 1;
schema.getField(1).uid = 2;
schema.getField(2).uid = 3;
- LogicalRelationalOperator l1 = new LOLoad(null, schema, plan, null);
+ LogicalRelationalOperator l1 = new LOLoad(schema, plan);
l1.setAlias("A");
plan.add(l1);
@@ -91,7 +91,7 @@ public class TestNewPlanFilterRule {
schema.getField(0).uid = 4;
schema.getField(1).uid = 5;
schema.getField(2).uid = 6;
- LogicalRelationalOperator l2 = new LOLoad(null, schema, plan, null);
+ LogicalRelationalOperator l2 = new LOLoad(schema, plan);
l2.setAlias("B");
plan.add(l2);
@@ -125,7 +125,7 @@ public class TestNewPlanFilterRule {
f1.setAlias("D");
plan.add(f1);
- LogicalRelationalOperator s1 = new LOStore(plan);
+ LogicalRelationalOperator s1 = new LOStore(plan, null, null, null);
plan.add(s1);
// load --|-join - filter - store
Modified: pig/trunk/test/org/apache/pig/test/TestNewPlanListener.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestNewPlanListener.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestNewPlanListener.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestNewPlanListener.java Tue Jun 19 00:33:11 2012
@@ -79,7 +79,7 @@ public class TestNewPlanListener extends
aschema.addField(new LogicalSchema.LogicalFieldSchema(
"x", null, DataType.INTEGER));
aschema.getField(0).uid = 1;
- LOLoad A = new LOLoad(null, null, lp, null);
+ LOLoad A = new LOLoad(null, lp);
A.neverUseForRealSetSchema(aschema);
lp.add(A);
@@ -88,7 +88,7 @@ public class TestNewPlanListener extends
bschema.addField(new LogicalSchema.LogicalFieldSchema(
"y", null, DataType.INTEGER));
bschema.getField(0).uid = 2;
- LOLoad B = new LOLoad(null, null, lp, null);
+ LOLoad B = new LOLoad(null, lp);
B.neverUseForRealSetSchema(bschema);
lp.add(B);
Modified: pig/trunk/test/org/apache/pig/test/TestNewPlanLogicalOptimizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestNewPlanLogicalOptimizer.java?rev=1351547&r1=1351546&r2=1351547&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestNewPlanLogicalOptimizer.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestNewPlanLogicalOptimizer.java Tue Jun 19 00:33:11 2012
@@ -17,6 +17,8 @@
*/
package org.apache.pig.test;
+import static org.apache.pig.newplan.logical.relational.LOTestHelper.newLOLoad;
+
import static org.junit.Assert.assertTrue;
import java.io.IOException;
@@ -84,7 +86,7 @@ public class TestNewPlanLogicalOptimizer
"x", null, DataType.BYTEARRAY));
aschema.addField(new LogicalSchema.LogicalFieldSchema(
"y", null, DataType.BYTEARRAY));
- LOLoad A = new LOLoad(new FileSpec("bla", new FuncSpec("PigStorage", "\t")), aschema, lp, conf);
+ LOLoad A = newLOLoad(new FileSpec("bla", new FuncSpec("PigStorage", "\t")), aschema, lp, conf);
A.setAlias("A");
lp.add(A);
@@ -94,7 +96,7 @@ public class TestNewPlanLogicalOptimizer
"a", null, DataType.BYTEARRAY));
bschema.addField(new LogicalSchema.LogicalFieldSchema(
"b", null, DataType.BYTEARRAY));
- LOLoad B = new LOLoad(new FileSpec("morebla", new FuncSpec("PigStorage", "\t")), bschema, lp, conf);
+ LOLoad B = newLOLoad(new FileSpec("morebla", new FuncSpec("PigStorage", "\t")), bschema, lp, conf);
B.setAlias("B");
lp.add(B);
@@ -160,7 +162,7 @@ public class TestNewPlanLogicalOptimizer
"x", null, DataType.BYTEARRAY));
aschema.addField(new LogicalSchema.LogicalFieldSchema(
"y", null, DataType.BYTEARRAY));
- LOLoad A = new LOLoad(new FileSpec("bla", new FuncSpec("PigStorage", "\t")), aschema, expected, conf);
+ LOLoad A = newLOLoad(new FileSpec("bla", new FuncSpec("PigStorage", "\t")), aschema, expected, conf);
expected.add(A);
// DA = filter
@@ -186,7 +188,7 @@ public class TestNewPlanLogicalOptimizer
"a", null, DataType.BYTEARRAY));
bschema.addField(new LogicalSchema.LogicalFieldSchema(
"b", null, DataType.BYTEARRAY));
- LOLoad B = new LOLoad(new FileSpec("morebla", new FuncSpec("PigStorage", "\t")), bschema, expected, conf);
+ LOLoad B = newLOLoad(new FileSpec("morebla", new FuncSpec("PigStorage", "\t")), bschema, expected, conf);
expected.add(B);
// DB = filter