You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ro...@apache.org on 2014/07/29 06:58:49 UTC
svn commit: r1614250 - in /pig/trunk: ./ conf/
shims/test/hadoop23/org/apache/pig/test/ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/...
Author: rohini
Date: Tue Jul 29 04:58:49 2014
New Revision: 1614250
URL: http://svn.apache.org/r1614250
Log:
PIG-4078: Port more mini cluster tests to Tez - part 6 (rohini)
Modified:
pig/trunk/CHANGES.txt
pig/trunk/build.xml
pig/trunk/conf/test-log4j.properties
pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java
pig/trunk/src/org/apache/pig/Main.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/tez/TezLauncher.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezSessionManager.java
pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java
pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java
pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java
pig/trunk/test/org/apache/pig/test/TestProjectRange.java
pig/trunk/test/org/apache/pig/test/TestScriptLanguage.java
pig/trunk/test/tez-tests
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Tue Jul 29 04:58:49 2014
@@ -50,6 +50,8 @@ OPTIMIZATIONS
BUG FIXES
+PIG-4078: Port more mini cluster tests to Tez - part 6 (rohini)
+
PIG-4071: Fix TestStore.testSetStoreSchema, TestParamSubPreproc.testGruntWithParamSub,
TestJobSubmission.testReducerNumEstimation (daijy)
Modified: pig/trunk/build.xml
URL: http://svn.apache.org/viewvc/pig/trunk/build.xml?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/build.xml (original)
+++ pig/trunk/build.xml Tue Jul 29 04:58:49 2014
@@ -151,8 +151,10 @@
</condition>
<target name="setTezEnv">
+ <propertyreset name="test.timeout" value="900000" />
<propertyreset name="hadoopversion" value="23" />
<propertyreset name="isHadoop23" value="true" />
+ <propertyreset name="hbase.hadoop.version" value="hadoop2" />
<propertyreset name="src.shims.dir" value="${basedir}/shims/src/hadoop${hadoopversion}" />
<propertyreset name="src.shims.test.dir" value="${basedir}/shims/test/hadoop${hadoopversion}" />
<propertyreset name="src.exclude.dir" value="" />
Modified: pig/trunk/conf/test-log4j.properties
URL: http://svn.apache.org/viewvc/pig/trunk/conf/test-log4j.properties?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/conf/test-log4j.properties (original)
+++ pig/trunk/conf/test-log4j.properties Tue Jul 29 04:58:49 2014
@@ -21,4 +21,4 @@ log4j.appender.A1=org.apache.log4j.Conso
# A1 uses PatternLayout.
log4j.appender.A1.layout=org.apache.log4j.PatternLayout
-log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+log4j.appender.A1.layout.ConversionPattern=%d [%t] %-5p %c %x - %m%n
Modified: pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java
URL: http://svn.apache.org/viewvc/pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java (original)
+++ pig/trunk/shims/test/hadoop23/org/apache/pig/test/TezMiniCluster.java Tue Jul 29 04:58:49 2014
@@ -21,12 +21,14 @@ import java.io.File;
import java.io.FileFilter;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.util.Map.Entry;
import org.apache.commons.io.filefilter.RegexFileFilter;
+import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.v2.MiniMRYarnCluster;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.pig.ExecType;
@@ -34,6 +36,7 @@ import org.apache.pig.PigConfiguration;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
import org.apache.pig.backend.hadoop.executionengine.tez.TezExecType;
import org.apache.pig.backend.hadoop.executionengine.tez.TezSessionManager;
+import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
public class TezMiniCluster extends MiniGenericCluster {
@@ -43,6 +46,7 @@ public class TezMiniCluster extends Mini
private static final File CORE_CONF_FILE = new File(CONF_DIR, "core-site.xml");
private static final File HDFS_CONF_FILE = new File(CONF_DIR, "hdfs-site.xml");
private static final File MAPRED_CONF_FILE = new File(CONF_DIR, "mapred-site.xml");
+ private static final File YARN_CONF_FILE = new File(CONF_DIR, "yarn-site.xml");
private static final ExecType TEZ = new TezExecType();
protected MiniMRYarnCluster m_mr = null;
@@ -74,12 +78,18 @@ public class TezMiniCluster extends Mini
//Create user home directory
m_fileSys.mkdirs(m_fileSys.getWorkingDirectory());
- m_dfs_conf.writeXml(new FileOutputStream(HDFS_CONF_FILE));
- m_fileSys.copyFromLocalFile(
- new Path(HDFS_CONF_FILE.getAbsoluteFile().toString()),
- new Path("/pigtest/conf/hdfs-site.xml"));
- Job job = Job.getInstance(m_dfs_conf);
- job.addFileToClassPath(new Path("/pigtest/conf/hdfs-site.xml"));
+ // Write core-site.xml
+ Configuration core_site = new Configuration(false);
+ core_site.set(FileSystem.FS_DEFAULT_NAME_KEY, m_dfs_conf.get(FileSystem.FS_DEFAULT_NAME_KEY));
+ core_site.writeXml(new FileOutputStream(CORE_CONF_FILE));
+
+ Configuration hdfs_site = new Configuration(false);
+ for (Entry<String, String> conf : m_dfs_conf) {
+ if (ArrayUtils.contains(m_dfs_conf.getPropertySources(conf.getKey()), "programatically")) {
+ hdfs_site.set(conf.getKey(), m_dfs_conf.getRaw(conf.getKey()));
+ }
+ }
+ hdfs_site.writeXml(new FileOutputStream(HDFS_CONF_FILE));
// Build mini YARN cluster
m_mr = new MiniMRYarnCluster("PigMiniCluster", 2);
@@ -90,30 +100,29 @@ public class TezMiniCluster extends Mini
m_mr_conf.set(YarnConfiguration.YARN_APPLICATION_CLASSPATH,
System.getProperty("java.class.path"));
- m_mr_conf.writeXml(new FileOutputStream(MAPRED_CONF_FILE));
- m_fileSys.copyFromLocalFile(
- new Path(MAPRED_CONF_FILE.getAbsoluteFile().toString()),
- new Path("/pigtest/conf/mapred-site.xml"));
- job.addFileToClassPath(new Path("/pigtest/conf/mapred-site.xml"));
+ Configuration mapred_site = new Configuration(false);
+ Configuration yarn_site = new Configuration(false);
+ for (Entry<String, String> conf : m_mr_conf) {
+ if (ArrayUtils.contains(m_mr_conf.getPropertySources(conf.getKey()), "programatically")) {
+ if (conf.getKey().contains("yarn")) {
+ yarn_site.set(conf.getKey(), m_mr_conf.getRaw(conf.getKey()));
+ } else if (!conf.getKey().startsWith("dfs")){
+ mapred_site.set(conf.getKey(), m_mr_conf.getRaw(conf.getKey()));
+ }
+ }
+ }
- // Write core-site.xml
- m_conf = m_mr_conf;
- m_conf.writeXml(new FileOutputStream(CORE_CONF_FILE));
- m_fileSys.copyFromLocalFile(
- new Path(CORE_CONF_FILE.getAbsoluteFile().toString()),
- new Path("/pigtest/conf/core-site.xml"));
- job.addFileToClassPath(new Path("/pigtest/conf/core-site.xml"));
+ mapred_site.writeXml(new FileOutputStream(MAPRED_CONF_FILE));
+ yarn_site.writeXml(new FileOutputStream(YARN_CONF_FILE));
// Write tez-site.xml
Configuration tez_conf = new Configuration(false);
// TODO PIG-3659 - Remove this once memory management is fixed
tez_conf.set(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, "20");
tez_conf.set("tez.lib.uris", "hdfs:///tez,hdfs:///tez/lib");
+ // Set to a lower value so that tests don't get stuck for long because of 1 AM running at a time
+ tez_conf.set(TezConfiguration.TEZ_SESSION_AM_DAG_SUBMIT_TIMEOUT_SECS, "20");
tez_conf.writeXml(new FileOutputStream(TEZ_CONF_FILE));
- m_fileSys.copyFromLocalFile(
- new Path(TEZ_CONF_FILE.getAbsoluteFile().toString()),
- new Path("/pigtest/conf/tez-site.xml"));
- job.addFileToClassPath(new Path("/pigtest/conf/tez-site.xml"));
// Copy tez jars to hdfs
m_fileSys.mkdirs(new Path("/tez/lib"));
@@ -131,6 +140,7 @@ public class TezMiniCluster extends Mini
}
}
+ m_conf = m_mr_conf;
// Turn FetchOptimizer off so that we can actually test Tez
m_conf.set(PigConfiguration.OPT_FETCH, System.getProperty("test.opt.fetch", "false"));
@@ -169,5 +179,8 @@ public class TezMiniCluster extends Mini
if(MAPRED_CONF_FILE.exists()) {
MAPRED_CONF_FILE.delete();
}
+ if(YARN_CONF_FILE.exists()) {
+ YARN_CONF_FILE.delete();
+ }
}
}
Modified: pig/trunk/src/org/apache/pig/Main.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/Main.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/Main.java (original)
+++ pig/trunk/src/org/apache/pig/Main.java Tue Jul 29 04:58:49 2014
@@ -88,6 +88,16 @@ import org.apache.pig.tools.timer.Perfor
@InterfaceStability.Stable
public class Main {
+ static {
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+
+ @Override
+ public void run() {
+ FileLocalizer.deleteTempResourceFiles();
+ }
+ });
+ }
+
private final static Log log = LogFactory.getLog(Main.class);
private static final String LOG4J_CONF = "log4jconf";
@@ -652,7 +662,6 @@ public class Main {
if (deleteTempFiles) {
// clear temp files
FileLocalizer.deleteTempFiles();
- FileLocalizer.deleteTempResourceFiles();
}
PerformanceTimerFactory.getPerfTimerFactory().dumpTimers();
}
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=1614250&r1=1614249&r2=1614250&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 Jul 29 04:58:49 2014
@@ -98,20 +98,22 @@ public abstract class HExecutionEngine i
return jc;
}
+ @Override
public DataStorage getDataStorage() {
return this.ds;
}
+ @Override
public void init() throws ExecException {
init(this.pigContext.getProperties());
}
- public JobConf getLocalConf(Properties properties) {
+ public JobConf getLocalConf() {
JobConf jc = new JobConf(false);
+ jc.addResource(CORE_DEFAULT_SITE);
jc.addResource(MAPRED_DEFAULT_SITE);
jc.addResource(YARN_DEFAULT_SITE);
- jc.addResource(CORE_DEFAULT_SITE);
return jc;
}
@@ -181,7 +183,7 @@ public abstract class HExecutionEngine i
properties.setProperty(FILE_SYSTEM_LOCATION, "file:///");
properties.setProperty(ALTERNATIVE_FILE_SYSTEM_LOCATION, "file:///");
- jc = getLocalConf(properties);
+ jc = getLocalConf();
}
// the method below alters the properties object by overriding the
@@ -260,7 +262,8 @@ public abstract class HExecutionEngine i
}
return result;
}
-
+
+ @Override
public PigStats launchPig(LogicalPlan lp, String grpName, PigContext pc)
throws FrontendException, ExecException {
@@ -285,6 +288,7 @@ public abstract class HExecutionEngine i
}
}
+ @Override
public void explain(LogicalPlan lp, PigContext pc, PrintStream ps,
String format, boolean verbose, File file, String suffix)
throws PlanException, VisitorException, IOException,
@@ -318,25 +322,30 @@ public abstract class HExecutionEngine i
}
}
+ @Override
public Properties getConfiguration() {
Properties properties = new Properties();
properties.putAll(pigContext.getProperties());
return properties;
}
+ @Override
public void setConfiguration(Properties newConfiguration) throws ExecException {
init(newConfiguration);
}
+ @Override
public void setProperty(String property, String value) {
Properties properties = pigContext.getProperties();
properties.put(property, value);
}
+ @Override
public ExecutableManager getExecutableManager() {
return new HadoopExecutableManager();
}
+ @Override
public void killJob(String jobID) throws BackendException {
if (launcher != null) {
launcher.killJob(jobID, getJobConf());
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=1614250&r1=1614249&r2=1614250&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 Jul 29 04:58:49 2014
@@ -155,7 +155,7 @@ public class MapReduceLauncher extends L
MRExecutionEngine exe = (MRExecutionEngine) pc.getExecutionEngine();
Properties defaultProperties = new Properties();
- JobConf defaultJobConf = exe.getLocalConf(defaultProperties);
+ JobConf defaultJobConf = exe.getLocalConf();
Utils.recomputeProperties(defaultJobConf, defaultProperties);
// This is a generic JobClient for checking progress of the jobs
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezLauncher.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezLauncher.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezLauncher.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezLauncher.java Tue Jul 29 04:58:49 2014
@@ -93,6 +93,7 @@ public class TezLauncher extends Launche
tezStats = new TezStats(pc);
PigStats.start(tezStats);
+ conf.set(TezConfiguration.TEZ_USE_CLUSTER_HADOOP_LIBS, "true");
TezJobCompiler jc = new TezJobCompiler(pc, conf);
TezPlanContainer tezPlanContainer = compile(php, pc);
@@ -328,7 +329,7 @@ public class TezLauncher extends Launche
if (!pc.inExplain && !pc.inDumpSchema) {
LoaderProcessor loaderStorer = new LoaderProcessor(tezPlan, pc);
loaderStorer.visit();
-
+
ParallelismSetter parallelismSetter = new ParallelismSetter(tezPlan, pc);
parallelismSetter.visit();
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezSessionManager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezSessionManager.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezSessionManager.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/TezSessionManager.java Tue Jul 29 04:58:49 2014
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -51,6 +52,9 @@ public class TezSessionManager {
});
}
+ private static ReentrantReadWriteLock sessionPoolLock = new ReentrantReadWriteLock();
+ private static boolean shutdown = false;
+
private TezSessionManager() {
}
@@ -98,69 +102,123 @@ public class TezSessionManager {
}
static TezClient getClient(Configuration conf, Map<String, LocalResource> requestedAMResources, Credentials creds) throws TezException, IOException {
- synchronized (sessionPool) {
- List<SessionInfo> sessionsToRemove = new ArrayList<SessionInfo>();
+ List<SessionInfo> sessionsToRemove = new ArrayList<SessionInfo>();
+ SessionInfo newSession = null;
+ sessionPoolLock.readLock().lock();
+ try {
+ if (shutdown == true) {
+ throw new IOException("TezSessionManager is shut down");
+ }
+
for (SessionInfo sessionInfo : sessionPool) {
- TezAppMasterStatus appMasterStatus = sessionInfo.session.getAppMasterStatus();
- if (appMasterStatus.equals(TezAppMasterStatus.SHUTDOWN)) {
- sessionsToRemove.add(sessionInfo);
- } else if (!sessionInfo.inUse && appMasterStatus.equals(TezAppMasterStatus.READY) &&
- validateSessionResources(sessionInfo, requestedAMResources)) {
- sessionInfo.inUse = true;
- return sessionInfo.session;
+ synchronized (sessionInfo) {
+ TezAppMasterStatus appMasterStatus = sessionInfo.session
+ .getAppMasterStatus();
+ if (appMasterStatus.equals(TezAppMasterStatus.SHUTDOWN)) {
+ sessionsToRemove.add(sessionInfo);
+ } else if (!sessionInfo.inUse
+ && appMasterStatus.equals(TezAppMasterStatus.READY)
+ && validateSessionResources(sessionInfo,requestedAMResources)) {
+ sessionInfo.inUse = true;
+ return sessionInfo.session;
+ }
}
}
-
+ } finally {
+ sessionPoolLock.readLock().unlock();
+ }
+ // We cannot find available AM, create new one
+ // Create session outside of locks so that getClient/freeSession is not
+ // blocked for parallel embedded pig runs
+ newSession = createSession(conf, requestedAMResources, creds);
+ newSession.inUse = true;
+ sessionPoolLock.writeLock().lock();
+ try {
+ if (shutdown == true) {
+ newSession.session.stop();
+ throw new IOException("TezSessionManager is shut down");
+ }
+ sessionPool.add(newSession);
for (SessionInfo sessionToRemove : sessionsToRemove) {
sessionPool.remove(sessionToRemove);
}
-
- // We cannot find available AM, create new one
- SessionInfo sessionInfo = createSession(conf, requestedAMResources, creds);
- sessionInfo.inUse = true;
- sessionPool.add(sessionInfo);
- return sessionInfo.session;
+ return newSession.session;
+ } finally {
+ sessionPoolLock.writeLock().unlock();
}
}
static void freeSession(TezClient session) {
- synchronized (sessionPool) {
+ sessionPoolLock.readLock().lock();
+ try {
for (SessionInfo sessionInfo : sessionPool) {
- if (sessionInfo.session == session) {
- sessionInfo.inUse = false;
- break;
+ synchronized (sessionInfo) {
+ if (sessionInfo.session == session) {
+ sessionInfo.inUse = false;
+ break;
+ }
}
}
+ } finally {
+ sessionPoolLock.readLock().unlock();
}
}
static void stopSession(TezClient session) throws TezException, IOException {
- synchronized (sessionPool) {
- Iterator<SessionInfo> iter = sessionPool.iterator();
+ Iterator<SessionInfo> iter = sessionPool.iterator();
+ SessionInfo sessionToRemove = null;
+ sessionPoolLock.readLock().lock();
+ try {
while (iter.hasNext()) {
SessionInfo sessionInfo = iter.next();
- if (sessionInfo.session == session) {
- log.info("Shutting down Tez session " + session);
- session.stop();
- iter.remove();
- break;
+ synchronized (sessionInfo) {
+ if (sessionInfo.session == session) {
+ log.info("Stopping Tez session " + session);
+ session.stop();
+ sessionToRemove = sessionInfo;
+ break;
+ }
}
}
+ } finally {
+ sessionPoolLock.readLock().unlock();
+ }
+ if (sessionToRemove != null) {
+ sessionPoolLock.writeLock().lock();
+ try {
+ sessionPool.remove(sessionToRemove);
+ } finally {
+ sessionPoolLock.writeLock().unlock();
+ }
}
}
@VisibleForTesting
public static void shutdown() {
- synchronized (sessionPool) {
+ try {
+ sessionPoolLock.writeLock().lock();
+ shutdown = true;
for (SessionInfo sessionInfo : sessionPool) {
- try {
- log.info("Shutting down Tez session " + sessionInfo.session);
- sessionInfo.session.stop();
- } catch (Exception e) {
- log.error("Error shutting down Tez session " + sessionInfo.session, e);
+ synchronized (sessionInfo) {
+ try {
+ if (sessionInfo.session.getAppMasterStatus().equals(
+ TezAppMasterStatus.SHUTDOWN)) {
+ log.info("Tez session is already shutdown "
+ + sessionInfo.session);
+ continue;
+ }
+ log.info("Shutting down Tez session "
+ + sessionInfo.session);
+ sessionInfo.session.stop();
+ } catch (Exception e) {
+ log.error("Error shutting down Tez session "
+ + sessionInfo.session, e);
+ }
}
}
sessionPool.clear();
+ } finally {
+ sessionPoolLock.writeLock().unlock();
}
}
}
Modified: pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java (original)
+++ pig/trunk/src/org/apache/pig/impl/io/FileLocalizer.java Tue Jul 29 04:58:49 2014
@@ -440,9 +440,7 @@ public class FileLocalizer {
new ThreadLocal<ContainerDescriptor>() {
};
- private static ThreadLocal<ContainerDescriptor> resourcePath =
- new ThreadLocal<ContainerDescriptor>() {
- };
+ private static ContainerDescriptor resourcePath;
/**
* This method is only used by test code to reset state.
@@ -485,10 +483,10 @@ public class FileLocalizer {
*/
public static synchronized ContainerDescriptor getTemporaryResourcePath(final PigContext pigContext)
throws DataStorageException {
- if (resourcePath.get() == null) {
- resourcePath.set(getTempContainer(pigContext));
+ if (resourcePath == null) {
+ resourcePath = getTempContainer(pigContext);
}
- return resourcePath.get();
+ return resourcePath;
}
private static synchronized ContainerDescriptor getTempContainer(final PigContext pigContext)
@@ -534,9 +532,9 @@ public class FileLocalizer {
}
public static void deleteTempResourceFiles() {
- if (resourcePath.get() != null) {
+ if (resourcePath != null) {
try {
- resourcePath.get().delete();
+ resourcePath.delete();
} catch (IOException e) {
log.error(e);
}
Modified: pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestCollectedGroup.java Tue Jul 29 04:58:49 2014
@@ -17,31 +17,32 @@
*/
package org.apache.pig.test;
-import java.io.*;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
import java.util.Iterator;
import java.util.List;
-import junit.framework.Assert;
-
import org.apache.pig.CollectableLoadFunc;
-import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.data.BagFactory;
-import org.apache.pig.data.DataBag;
-import org.apache.pig.data.Tuple;
-import org.apache.pig.test.utils.TestHelper;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRCompilerException;
-import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCollectedGroup;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCollectedGroup;
import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.plan.OperatorKey;
import org.apache.pig.newplan.logical.relational.LOCogroup;
import org.apache.pig.newplan.logical.relational.LOStore;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
+import org.apache.pig.test.utils.TestHelper;
import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -50,14 +51,14 @@ import org.junit.runners.JUnit4;
@RunWith(JUnit4.class)
public class TestCollectedGroup {
private static final String INPUT_FILE = "MapSideGroupInput.txt";
-
+
private PigServer pigServer;
- private static MiniCluster cluster = MiniCluster.buildCluster();
+ private static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
public TestCollectedGroup() throws ExecException, IOException{
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
}
-
+
@Before
public void setUp() throws Exception {
createFiles();
@@ -70,29 +71,29 @@ public class TestCollectedGroup {
w.println("100\tapple2\t74");
w.println("200\torange1\t100");
w.println("200\torange2\t89");
- w.println("300\tstrawberry\t64");
- w.println("300\tstrawberry\t64");
- w.println("300\tstrawberry\t76");
+ w.println("300\tstrawberry\t64");
+ w.println("300\tstrawberry\t64");
+ w.println("300\tstrawberry\t76");
w.println("400\tpear\t78");
w.close();
-
+
Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
}
-
+
@After
public void tearDown() throws Exception {
new File(INPUT_FILE).delete();
Util.deleteFile(cluster, INPUT_FILE);
}
-
+
@Test
public void testNonCollectableLoader() throws Exception{
String query = "A = LOAD '" + INPUT_FILE + "' as (id, name, grade);" +
"B = group A by id using 'collected';";
- PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
+ PigContext pc = new PigContext(cluster.getExecType(),cluster.getProperties());
pc.connect();
try {
- Util.buildMRPlan(Util.buildPp(pigServer, query),pc);
+ Util.buildMRPlan(Util.buildPp(pigServer, query),pc);
Assert.fail("Must throw MRCompiler Exception");
} catch (Exception e) {
Assert.assertTrue(e instanceof MRCompilerException);
@@ -101,7 +102,7 @@ public class TestCollectedGroup {
@Test
public void testCollectedGrpSpecifiedInSingleQuotes1() throws Exception {
- String query = "A = LOAD '" + INPUT_FILE + "' as (id, name, grade);" +
+ String query = "A = LOAD '" + INPUT_FILE + "' as (id, name, grade);" +
"B = group A by id using 'collected';" +
"Store B into 'y';";
LogicalPlan lp = Util.buildLp(pigServer, query );
@@ -109,7 +110,7 @@ public class TestCollectedGroup {
LOCogroup grp = (LOCogroup)lp.getPredecessors( store ).get(0);
Assert.assertEquals( LOCogroup.GROUPTYPE.COLLECTED, grp.getGroupType() );
}
-
+
@Test
public void testCollectedGrpSpecifiedInSingleQuotes2() throws Exception{
String query = "A = LOAD '" + INPUT_FILE + "' as (id, name, grade);" +
@@ -120,12 +121,12 @@ public class TestCollectedGroup {
LOCogroup grp = (LOCogroup)lp.getPredecessors( store ).get(0);
Assert.assertEquals(LOCogroup.GROUPTYPE.REGULAR, grp.getGroupType());
}
-
+
@AfterClass
public static void oneTimeTearDown() throws Exception {
cluster.shutDown();
}
-
+
@Test
public void testPOMapsideGroupNoNullPlans() throws IOException {
POCollectedGroup pmg = new POCollectedGroup(new OperatorKey());
@@ -133,15 +134,15 @@ public class TestCollectedGroup {
Assert.assertTrue(plans != null);
Assert.assertTrue(plans.size() == 0);
- }
-
- @Test
+ }
+
+ @Test
public void testMapsideGroupParserNoSupportForMultipleInputs() throws IOException {
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (id, name, grade);");
pigServer.registerQuery("B = LOAD '" + INPUT_FILE + "' as (id, name, grade);");
-
+
try {
pigServer.registerQuery("C = group A by id, B by id using 'collected';");
pigServer.openIterator( "C" );
@@ -151,13 +152,13 @@ public class TestCollectedGroup {
Assert.assertTrue( e.getMessage().contains( msg ) );
}
}
-
+
@Test
public void testMapsideGroupParserNoSupportForGroupAll() throws IOException {
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (id, name, grade);");
-
+
try {
pigServer.registerQuery("B = group A all using 'collected';");
pigServer.openIterator( "B" );
@@ -167,13 +168,13 @@ public class TestCollectedGroup {
Assert.assertTrue( e.getMessage().contains( msg ) );
}
}
-
+
@Test
public void testMapsideGroupParserNoSupportForByExpression() throws IOException {
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' as (id, name, grade);");
-
+
try {
pigServer.registerQuery("B = group A by id*grade using 'collected';");
pigServer.openIterator("B");
@@ -186,7 +187,7 @@ public class TestCollectedGroup {
@Test
public void testMapsideGroupByOneColumn() throws IOException{
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' using "+DummyCollectableLoader.class.getName() +"() as (id, name, grade);");
try {
@@ -218,13 +219,13 @@ public class TestCollectedGroup {
Assert.fail(e.getMessage());
}
}
-
+
@Test
public void testMapsideGroupByMultipleColumns() throws IOException{
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' using "+DummyCollectableLoader.class.getName() +"() as (id, name, grade);");
-
+
try {
DataBag dbfrj = BagFactory.getInstance().newDefaultBag();
DataBag dbshj = BagFactory.getInstance().newDefaultBag();
@@ -254,13 +255,13 @@ public class TestCollectedGroup {
Assert.fail(e.getMessage());
}
}
-
+
@Test
public void testMapsideGroupByStar() throws IOException{
- pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
pigServer.registerQuery("A = LOAD '" + INPUT_FILE + "' using "+DummyCollectableLoader.class.getName() +"() as (id, name, grade);");
-
+
try {
DataBag dbfrj = BagFactory.getInstance().newDefaultBag();
DataBag dbshj = BagFactory.getInstance().newDefaultBag();
@@ -296,6 +297,6 @@ public class TestCollectedGroup {
@Override
public void ensureAllKeyInstancesInSameSplit() throws IOException {
}
-
+
}
}
Modified: pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestEvalPipeline2.java Tue Jul 29 04:58:49 2014
@@ -1051,7 +1051,6 @@ public class TestEvalPipeline2 {
// See PIG-1787
@Test
public void testOrderByLimitJoin() throws Exception{
- Assume.assumeTrue("Skip this test for TEZ till TEZ-1257 is fixed", Util.isMapredExecType(cluster.getExecType()));
String[] input1 = {
"1\t1",
"1\t2"
@@ -1579,7 +1578,6 @@ public class TestEvalPipeline2 {
// See PIG-2237
@Test
public void testLimitAutoReducer() throws Exception{
- Assume.assumeTrue("Skip this test for TEZ till TEZ-1257 is fixed", Util.isMapredExecType(cluster.getExecType()));
String[] input = {
"1\tA",
"4\tB",
Modified: pig/trunk/test/org/apache/pig/test/TestProjectRange.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestProjectRange.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestProjectRange.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestProjectRange.java Tue Jul 29 04:58:49 2014
@@ -18,8 +18,6 @@
package org.apache.pig.test;
-import static org.apache.pig.ExecType.LOCAL;
-import static org.apache.pig.ExecType.MAPREDUCE;
import static org.junit.Assert.assertEquals;
import java.io.File;
@@ -27,35 +25,26 @@ import java.io.IOException;
import java.util.Iterator;
import java.util.List;
-import junit.framework.Assert;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.pig.ExecType;
-import org.apache.pig.ExecTypeProvider;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
-import org.apache.pig.impl.plan.CompilationMessageCollector;
import org.apache.pig.impl.util.Utils;
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.relational.LOCogroup;
import org.apache.pig.newplan.logical.relational.LOJoin;
import org.apache.pig.newplan.logical.relational.LOSort;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
-import org.apache.pig.newplan.logical.visitor.CastLineageSetter;
-import org.apache.pig.newplan.logical.visitor.ColumnAliasConversionVisitor;
-import org.apache.pig.newplan.logical.visitor.SchemaAliasVisitor;
-import org.apache.pig.newplan.logical.visitor.TypeCheckingRelVisitor;
-import org.apache.pig.newplan.logical.visitor.UnionOnSchemaSetter;
import org.apache.pig.parser.ParserException;
import org.apache.pig.parser.ParserTestingUtils;
import org.apache.pig.test.utils.NewLogicalPlanUtil;
import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -64,38 +53,21 @@ public class TestProjectRange {
protected final Log log = LogFactory.getLog(getClass());
- protected static ExecType execType = LOCAL;
- private static MiniCluster cluster;
+ private static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
protected static PigServer pigServer;
private static final String INP_FILE_5FIELDS = "TestProjectRange_5fields";
@BeforeClass
public static void oneTimeSetUp() throws Exception {
-
- String execTypeString = System.getProperty("test.exectype");
-
- if(execTypeString!=null && execTypeString.length()>0){
- execType = ExecTypeProvider.fromString(execTypeString);
- }
-
String[] input = {"10\t20\t30\t40\t50", "11\t21\t31\t41\t51"};
- Util.createLocalInputFile(INP_FILE_5FIELDS, input);
-
- if(execType == MAPREDUCE) {
- cluster = MiniCluster.buildCluster();
- Util.copyFromLocalToCluster(cluster, INP_FILE_5FIELDS, INP_FILE_5FIELDS);
- }
+ Util.createInputFile(cluster, INP_FILE_5FIELDS, input);
}
@Before
- public void setup() throws ExecException{
- if(execType == MAPREDUCE) {
- pigServer = new PigServer(MAPREDUCE, cluster.getProperties());
- } else {
- pigServer = new PigServer(LOCAL);
- }
+ public void setup() throws ExecException {
+ pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
}
@After
@@ -1087,11 +1059,11 @@ public class TestProjectRange {
List<Tuple> expectedRes =
Util.getTuplesFromConstantTupleStrings(
new String[] {
- "((30,40,50),{(10,20,30,40,50)})",
- "((31,41,51),{(11,21,31,41,51)})",
+ "((30,40,50),{(10,20L,30,40,50)})",
+ "((31,41,51),{(11,21L,31,41,51)})",
});
Iterator<Tuple> it = pigServer.openIterator("lim");
- Util.checkQueryOutputs(it, expectedRes);
+ Util.checkQueryOutputsAfterSort(it, expectedRes);
}
Modified: pig/trunk/test/org/apache/pig/test/TestScriptLanguage.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestScriptLanguage.java?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestScriptLanguage.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestScriptLanguage.java Tue Jul 29 04:58:49 2014
@@ -28,7 +28,6 @@ import java.util.List;
import java.util.Map;
import org.apache.hadoop.fs.Path;
-import org.apache.pig.ExecType;
import org.apache.pig.PigRunner;
import org.apache.pig.PigServer;
import org.apache.pig.data.Tuple;
@@ -38,12 +37,13 @@ import org.apache.pig.tools.pigstats.Pig
import org.apache.pig.tools.pigstats.PigStatsUtil;
import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
public class TestScriptLanguage {
- static MiniCluster cluster = MiniCluster.buildCluster();
+ static MiniGenericCluster cluster = MiniGenericCluster.buildCluster();
@BeforeClass
public static void setUpBeforeClass() throws Exception {
@@ -71,7 +71,7 @@ public class TestScriptLanguage {
String scriptName = name + "_testScript.py";
Util.createLocalInputFile(scriptName, script);
ScriptEngine scriptEngine = ScriptEngine.getInstance("jython");
- PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ PigServer pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
Map<String, List<PigStats>> statsMap = scriptEngine.run(pigServer.getPigContext(), scriptName);
return statsMap;
}
@@ -88,7 +88,7 @@ public class TestScriptLanguage {
private PigStats runPigRunner(String name, String[] script, boolean shouldSucceed) throws Exception {
String scriptName = createScript(name, script);
- String[] args = { "-g", "jython", scriptName };
+ String[] args = { "-x", cluster.getExecType().name().toLowerCase(), "-g", "jython", scriptName };
PigStats mainStats = PigRunner.run(args, new TestPigRunner.TestNotificationListener());
if (shouldSucceed) {
@@ -599,7 +599,7 @@ public class TestScriptLanguage {
// See PIG-2291
@Test
public void testIllustrateInScript() throws Exception {
-
+ Assume.assumeTrue("Skip this test for TEZ. See PIG-3993", Util.isMapredExecType(cluster.getExecType()));
String[] script = { "#!/usr/bin/python",
"from org.apache.pig.scripting import *",
"Pig.fs(\"rmr simple_out\")",
@@ -634,7 +634,7 @@ public class TestScriptLanguage {
assertEquals(0, statsMap.size());
}
-
+
@Test
public void testSysArguments() throws Exception {
String[] script = {
@@ -649,32 +649,32 @@ public class TestScriptLanguage {
String file1 = "/tmp/sysarg_1";
String file2 = "/tmp/sysarg_2";
createEmptyFiles(file1, file2);
-
+
File scriptFile = Util.createInputFile("sysarg", ".py", script);
// ExecMode.STRING
PigStats stats = PigRunner.run(new String[] { scriptFile.getAbsolutePath(),
file1 + "," + file2 }, null);
assertEquals(null, stats.getErrorMessage());
assertFileNotExists(file1, file2);
-
+
createEmptyFiles(file1, file2);
// Clear stats from previous execution
PigStatsUtil.getEmptyPigStats();
-
+
// ExecMode.FILE
- stats = PigRunner.run(new String[] { "-f", scriptFile.getAbsolutePath(), "arg0",
+ stats = PigRunner.run(new String[] { "-f", scriptFile.getAbsolutePath(), "arg0",
file1 + "," + file2 }, null);
assertEquals(null, stats.getErrorMessage());
assertFileNotExists(file1, file2);
}
-
+
private void createEmptyFiles(String... filenames) throws IOException {
for (String file : filenames) {
Util.createInputFile(cluster, file, new String[]{""});
assertTrue(cluster.getFileSystem().exists(new Path(file)));
}
}
-
+
private void assertFileNotExists(String... filenames) throws IOException {
for (String file : filenames) {
assertFalse(cluster.getFileSystem().exists(new Path(file)));
Modified: pig/trunk/test/tez-tests
URL: http://svn.apache.org/viewvc/pig/trunk/test/tez-tests?rev=1614250&r1=1614249&r2=1614250&view=diff
==============================================================================
--- pig/trunk/test/tez-tests (original)
+++ pig/trunk/test/tez-tests Tue Jul 29 04:58:49 2014
@@ -5,6 +5,7 @@
**/TestBinaryExpressionOps.java
**/TestBuiltin.java
**/TestCharArrayToNumeric.java
+**/TestCollectedGroup.java
**/TestCombiner.java
**/TestCompressedFiles.java
**/TestCustomPartitioner.java
@@ -35,10 +36,12 @@
**/TestPigServerWithMacros.java
**/TestPigSplit.java
**/TestPigStorage.java
+**/TestProjectRange.java
**/TestRegisteredJarVisibility.java
**/TestSample.java
**/TestScalarAliases.java
**/TestSchema.java
+**/TestScriptLanguage.java
**/TestScriptLanguageJavaScript.java
**/TestScriptUDF.java
**/TestSkewedJoin.java
@@ -51,7 +54,7 @@
**/TestUDF.java
**/TestUDFContext.java
**/TestSecondarySortTez.java
+**/TestTezAutoParallelism.java
**/TestTezCompiler.java
**/TestTezJobControlCompiler.java
**/TestTezLauncher.java
-**/TestTezAutoParallelism.java