You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ch...@apache.org on 2013/09/18 22:21:08 UTC
svn commit: r1524555 [2/3] - in /pig/branches/tez: ./ conf/ ivy/ src/
src/org/apache/pig/ src/org/apache/pig/backend/executionengine/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/ s...
Modified: pig/branches/tez/src/org/apache/pig/impl/streaming/OutputHandler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/streaming/OutputHandler.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/streaming/OutputHandler.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/streaming/OutputHandler.java Wed Sep 18 20:21:06 2013
@@ -21,16 +21,17 @@ import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.util.LineReader;
+import org.apache.pig.PigStreamingBase;
import org.apache.pig.StreamToPig;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
/**
- * {@link OutputHandler} is responsible for handling the output of the
+ * {@link OutputHandler} is responsible for handling the output of the
* Pig-Streaming external command.
- *
- * The output of the managed executable could be fetched in a
- * {@link OutputType#SYNCHRONOUS} manner via its <code>stdout</code> or in an
+ *
+ * The output of the managed executable could be fetched in a
+ * {@link OutputType#SYNCHRONOUS} manner via its <code>stdout</code> or in an
* {@link OutputType#ASYNCHRONOUS} manner via an external file to which the
* process wrote its output.
*/
@@ -39,30 +40,34 @@ public abstract class OutputHandler {
/*
* The deserializer to be used to send data to the managed process.
- *
+ *
* It is the responsibility of the concrete sub-classes to setup and
- * manage the deserializer.
- */
+ * manage the deserializer.
+ */
protected StreamToPig deserializer;
-
+
+ private PigStreamingBase newDeserializer;
+
protected LineReader in = null;
+ private Text currValue = new Text();
+
private BufferedPositionedInputStream istream;
-
+
/**
* Get the handled <code>OutputType</code>.
- * @return the handled <code>OutputType</code>
+ * @return the handled <code>OutputType</code>
*/
public abstract OutputType getOutputType();
-
+
// flag to mark if close() has already been called
protected boolean alreadyClosed = false;
-
+
/**
* Bind the <code>OutputHandler</code> to the <code>InputStream</code>
* from which to read the output data of the managed process.
- *
- * @param is <code>InputStream</code> from which to read the output data
+ *
+ * @param is <code>InputStream</code> from which to read the output data
* of the managed process
* @throws IOException
*/
@@ -70,11 +75,14 @@ public abstract class OutputHandler {
long offset, long end) throws IOException {
this.istream = is;
this.in = new LineReader(istream);
+ if (this.deserializer instanceof PigStreamingBase) {
+ this.newDeserializer = (PigStreamingBase) deserializer;
+ }
}
-
+
/**
* Get the next output <code>Tuple</code> of the managed process.
- *
+ *
* @return the next output <code>Tuple</code> of the managed process
* @throws IOException
*/
@@ -83,17 +91,21 @@ public abstract class OutputHandler {
return null;
}
- Text value = new Text();
- int num = in.readLine(value);
+ currValue.clear();
+ int num = in.readLine(currValue);
if (num <= 0) {
return null;
}
-
- byte[] newBytes = new byte[value.getLength()];
- System.arraycopy(value.getBytes(), 0, newBytes, 0, value.getLength());
- return deserializer.deserialize(newBytes);
+
+ if (newDeserializer != null) {
+ return newDeserializer.deserialize(currValue.getBytes(), 0, currValue.getLength());
+ } else {
+ byte[] newBytes = new byte[currValue.getLength()];
+ System.arraycopy(currValue.getBytes(), 0, newBytes, 0, currValue.getLength());
+ return deserializer.deserialize(newBytes);
+ }
}
-
+
/**
* Close the <code>OutputHandler</code>.
* @throws IOException
Modified: pig/branches/tez/src/org/apache/pig/impl/streaming/StreamingCommand.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/streaming/StreamingCommand.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/streaming/StreamingCommand.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/streaming/StreamingCommand.java Wed Sep 18 20:21:06 2013
@@ -34,6 +34,7 @@ import org.apache.pig.builtin.PigStorage
import org.apache.pig.builtin.PigStreaming;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.zookeeper.Shell;
/**
@@ -185,6 +186,10 @@ public class StreamingCommand implements
URI pathUri = null;
URI dfsPath = null;
try {
+ // On Windows, replace "\" into "/"
+ if (Shell.WINDOWS) {
+ path = path.replaceAll("\\\\", "/");
+ }
pathUri = new URI(path);
// Strip away the URI's _fragment_ and _query_
Modified: pig/branches/tez/src/org/apache/pig/impl/util/JavaCompilerHelper.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/util/JavaCompilerHelper.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/util/JavaCompilerHelper.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/util/JavaCompilerHelper.java Wed Sep 18 20:21:06 2013
@@ -33,6 +33,7 @@ import javax.tools.ToolProvider;
import com.google.common.collect.Lists;
+import org.apache.hadoop.util.Shell;
import org.apache.log4j.Logger;
public class JavaCompilerHelper {
@@ -85,7 +86,7 @@ public class JavaCompilerHelper {
}
public void addToClassPath(String path) {
- this.classPath = this.classPath+":"+path;
+ this.classPath = this.classPath + System.getProperty("path.separator") + path;
}
public String getClassPath() {
Modified: pig/branches/tez/src/org/apache/pig/impl/util/StorageUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/util/StorageUtil.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/util/StorageUtil.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/util/StorageUtil.java Wed Sep 18 20:21:06 2013
@@ -229,14 +229,24 @@ public final class StorageUtil {
* @return tuple constructed from the text
*/
public static Tuple textToTuple(Text val, byte fieldDel) {
+ return bytesToTuple(val.getBytes(), 0, val.getLength(), fieldDel);
+ }
+
+ /**
+ * Transform bytes from a byte array up to the specified length to a <code>Tuple</code>
+ *
+ * @param buf the byte array
+ * @param length number of bytes to consume from the byte array
+ * @param fieldDel the field delimiter
+ * @return tuple constructed from the bytes
+ */
+ public static Tuple bytesToTuple(byte[] buf, int offset, int length, byte fieldDel) {
- byte[] buf = val.getBytes();
- int len = val.getLength();
- int start = 0;
+ int start = offset;
ArrayList<Object> protoTuple = new ArrayList<Object>();
- for (int i = 0; i < len; i++) {
+ for (int i = offset; i < length; i++) {
if (buf[i] == fieldDel) {
readField(protoTuple, buf, start, i);
start = i + 1;
@@ -244,8 +254,8 @@ public final class StorageUtil {
}
// pick up the last field
- if (start <= len) {
- readField(protoTuple, buf, start, len);
+ if (start <= length) {
+ readField(protoTuple, buf, start, length);
}
return TupleFactory.getInstance().newTupleNoCopy(protoTuple);
Modified: pig/branches/tez/src/org/apache/pig/impl/util/UriUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/impl/util/UriUtil.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/impl/util/UriUtil.java (original)
+++ pig/branches/tez/src/org/apache/pig/impl/util/UriUtil.java Wed Sep 18 20:21:06 2013
@@ -17,6 +17,8 @@
*/
package org.apache.pig.impl.util;
+import java.io.File;
+
public class UriUtil {
public static boolean isHDFSFile(String uri){
@@ -31,7 +33,8 @@ public class UriUtil {
public static boolean isHDFSFileOrLocalOrS3N(String uri){
if(uri == null)
return false;
- if(uri.startsWith("/") || uri.startsWith("hdfs:") || uri.startsWith("file:") || uri.startsWith("s3n:")){
+ if(uri.startsWith("/") || uri.matches("[A-Za-z]:.*") || uri.startsWith("hdfs:")
+ || uri.startsWith("file:") || uri.startsWith("s3n:")){
return true;
}
return false;
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOLoad.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOLoad.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOLoad.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LOLoad.java Wed Sep 18 20:21:06 2013
@@ -51,6 +51,7 @@ public class LOLoad extends LogicalRelat
private final String schemaFile;
private final String signature;
private long limit = -1;
+ private boolean isTmpLoad;
/**
* used for pattern matching
@@ -280,6 +281,14 @@ public class LOLoad extends LogicalRelat
return signature;
}
+ public boolean isTmpLoad() {
+ return isTmpLoad;
+ }
+
+ public void setTmpLoad(boolean isTmpLoad) {
+ this.isTmpLoad = isTmpLoad;
+ }
+
public LogicalSchema getScriptSchema() {
return scriptSchema;
}
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java Wed Sep 18 20:21:06 2013
@@ -141,6 +141,8 @@ public class LogToPhyTranslationVisitor
load.setResultType(DataType.BAG);
load.setSignature(loLoad.getSignature());
load.setLimit(loLoad.getLimit());
+ load.setIsTmpLoad(loLoad.isTmpLoad());
+
currentPlan.add(load);
logToPhyMap.put(loLoad, load);
Modified: pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogicalPlan.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogicalPlan.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogicalPlan.java (original)
+++ pig/branches/tez/src/org/apache/pig/newplan/logical/relational/LogicalPlan.java Wed Sep 18 20:21:06 2013
@@ -18,40 +18,43 @@
package org.apache.pig.newplan.logical.relational;
-import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.util.HashOutputStream;
import org.apache.pig.newplan.BaseOperatorPlan;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.OperatorPlan;
import org.apache.pig.newplan.logical.DotLOPrinter;
import org.apache.pig.newplan.logical.optimizer.LogicalPlanPrinter;
+import com.google.common.hash.HashFunction;
+import com.google.common.hash.Hashing;
+
/**
- * LogicalPlan is the logical view of relational operations Pig will execute
+ * LogicalPlan is the logical view of relational operations Pig will execute
* for a given script. Note that it contains only relational operations.
* All expressions will be contained in LogicalExpressionPlans inside
* each relational operator.
*/
public class LogicalPlan extends BaseOperatorPlan {
-
+
public LogicalPlan(LogicalPlan other) {
// shallow copy constructor
super(other);
}
-
+
public LogicalPlan() {
super();
}
-
+
/**
* Equality is checked by calling equals on every leaf in the plan. This
- * assumes that plans are always connected graphs. It is somewhat
- * inefficient since every leaf will test equality all the way to
+ * assumes that plans are always connected graphs. It is somewhat
+ * inefficient since every leaf will test equality all the way to
* every root. But it is only intended for use in testing, so that
* should be ok. Checking predecessors (as opposed to successors) was
* chosen because splits (which have multiple successors) do not depend
@@ -60,25 +63,32 @@ public class LogicalPlan extends BaseOpe
* graph has no correctness implications, whereas reversing the inputs
* of join can. This method of doing equals will detect predecessors
* in different orders but not successors in different orders.
- * It will return false if either plan has non deterministic EvalFunc.
+ * It will return false if either plan has non deterministic EvalFunc.
*/
@Override
public boolean isEqual(OperatorPlan other) throws FrontendException {
if (other == null || !(other instanceof LogicalPlan)) {
return false;
}
-
- return super.isEqual(other);
+
+ return super.isEqual(other);
}
-
+
@Override
- public void explain(PrintStream ps, String format, boolean verbose)
+ public void explain(PrintStream ps, String format, boolean verbose)
throws FrontendException {
+ if (format.equals("xml")) {
+ ps.println("<logicalPlan>XML Not Supported</logicalPlan>");
+ return;
+ }
+
ps.println("#-----------------------------------------------");
ps.println("# New Logical Plan:");
ps.println("#-----------------------------------------------");
- if (format.equals("dot")) {
+ if (this.size() == 0) {
+ ps.println("Logical plan is empty.");
+ } else if (format.equals("dot")) {
DotLOPrinter lpp = new DotLOPrinter(this, ps);
lpp.dump();
} else {
@@ -98,7 +108,7 @@ public class LogicalPlan extends BaseOpe
ops.add( op );
}
}
-
+
if( ops.isEmpty() ) {
return null;
} else {
@@ -109,18 +119,21 @@ public class LogicalPlan extends BaseOpe
/**
* Returns the signature of the LogicalPlan. The signature is a unique identifier for a given
* plan generated by a Pig script. The same script run multiple times with the same version of
- * Pig is guarenteed to produce the same signature, even if the input or output locations differ.
+ * Pig is guaranteed to produce the same signature, even if the input or output locations differ.
*
* @return a unique identifier for the logical plan
* @throws FrontendException if signature can't be computed
*/
public String getSignature() throws FrontendException {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- PrintStream ps = new PrintStream(baos);
+
+ // Use a streaming hash function. goodFastHash(32) is murmur3 32 bits
+ HashFunction hf = Hashing.goodFastHash(32);
+ HashOutputStream hos = new HashOutputStream(hf);
+ PrintStream ps = new PrintStream(hos);
LogicalPlanPrinter printer = new LogicalPlanPrinter(this, ps);
printer.visit();
- return Integer.toString(baos.toString().hashCode());
+ return Integer.toString(hos.getHashCode().asInt());
}
}
Modified: pig/branches/tez/src/org/apache/pig/parser/DryRunGruntParser.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/parser/DryRunGruntParser.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/parser/DryRunGruntParser.java (original)
+++ pig/branches/tez/src/org/apache/pig/parser/DryRunGruntParser.java Wed Sep 18 20:21:06 2013
@@ -124,6 +124,9 @@ public class DryRunGruntParser extends P
if (isVerbose) sb.append("-brief ");
if (format != null && format.equals("dot")) {
sb.append("-dot ");
+ }
+ if (format != null && format.equals("xml")) {
+ sb.append("-xml ");
}
if (params != null) {
for (String param : params) {
Modified: pig/branches/tez/src/org/apache/pig/parser/LogicalPlanBuilder.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/parser/LogicalPlanBuilder.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/parser/LogicalPlanBuilder.java (original)
+++ pig/branches/tez/src/org/apache/pig/parser/LogicalPlanBuilder.java Wed Sep 18 20:21:06 2013
@@ -855,6 +855,7 @@ public class LogicalPlanBuilder {
ConfigurationUtil.toConfiguration(pigContext.getProperties()),
loFunc,
alias + "_" + newOperatorKey());
+ op.setTmpLoad(false);
return buildOp( loc, op, alias, new ArrayList<String>(), null );
}
Modified: pig/branches/tez/src/org/apache/pig/parser/ParserException.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/parser/ParserException.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/parser/ParserException.java (original)
+++ pig/branches/tez/src/org/apache/pig/parser/ParserException.java Wed Sep 18 20:21:06 2013
@@ -33,6 +33,10 @@ public class ParserException extends Fro
public ParserException(String msg) {
super( msg, errorCode );
}
+
+ public ParserException(String msg, SourceLocation location) {
+ super(msg, errorCode, location);
+ }
public ParserException(String msg, Throwable cause) {
super( msg, errorCode, cause );
Modified: pig/branches/tez/src/org/apache/pig/parser/QueryParserDriver.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/parser/QueryParserDriver.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/parser/QueryParserDriver.java (original)
+++ pig/branches/tez/src/org/apache/pig/parser/QueryParserDriver.java Wed Sep 18 20:21:06 2013
@@ -237,7 +237,8 @@ public class QueryParserDriver {
} catch (RecognitionException e) {
String msg = parser.getErrorHeader(e) + " "
+ parser.getErrorMessage(e, parser.getTokenNames());
- throw new ParserException(msg);
+ SourceLocation location = new SourceLocation(null, e.line,e.charPositionInLine);
+ throw new ParserException(msg, location);
} catch(RuntimeException ex) {
throw new ParserException( ex.getMessage() );
}
Modified: pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyEvalFunc.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyEvalFunc.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyEvalFunc.java (original)
+++ pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyEvalFunc.java Wed Sep 18 20:21:06 2013
@@ -21,6 +21,7 @@ package org.apache.pig.scripting.groovy;
import groovy.util.ResourceException;
import groovy.util.ScriptException;
+import java.io.File;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.lang.reflect.Constructor;
@@ -65,7 +66,7 @@ public class GroovyEvalFunc<T> extends E
if (null == c) {
try {
- c = GroovyScriptEngine.getEngine().loadScriptByName(path);
+ c = GroovyScriptEngine.getEngine().loadScriptByName(new File(path).toURI().toString());
} catch (ScriptException se) {
throw new IOException(se);
} catch (ResourceException re) {
Modified: pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyScriptEngine.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyScriptEngine.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyScriptEngine.java (original)
+++ pig/branches/tez/src/org/apache/pig/scripting/groovy/GroovyScriptEngine.java Wed Sep 18 20:21:06 2013
@@ -22,6 +22,7 @@ import groovy.lang.Tuple;
import groovy.util.ResourceException;
import groovy.util.ScriptException;
+import java.io.File;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
@@ -102,7 +103,7 @@ public class GroovyScriptEngine extends
// Load the script
//
- Class c = gse.loadScriptByName(scriptFile);
+ Class c = gse.loadScriptByName(new File(scriptFile).toURI().toString());
//
// Extract the main method
@@ -144,7 +145,7 @@ public class GroovyScriptEngine extends
// Read file
//
- Class c = gse.loadScriptByName(path);
+ Class c = gse.loadScriptByName(new File(path).toURI().toString());
//
// Keep track of initial/intermed/final methods of Albegraic UDFs
Modified: pig/branches/tez/src/org/apache/pig/tools/grunt/GruntParser.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/tools/grunt/GruntParser.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/tools/grunt/GruntParser.java (original)
+++ pig/branches/tez/src/org/apache/pig/tools/grunt/GruntParser.java Wed Sep 18 20:21:06 2013
@@ -46,7 +46,9 @@ import jline.ConsoleReaderInputStream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsShell;
+import org.apache.hadoop.fs.Path;
import org.apache.pig.LoadFunc;
import org.apache.pig.PigServer;
import org.apache.pig.backend.datastorage.ContainerDescriptor;
@@ -72,7 +74,6 @@ import org.apache.pig.tools.pigstats.Pig
import org.fusesource.jansi.Ansi;
import org.fusesource.jansi.AnsiConsole;
-@SuppressWarnings("deprecation")
public class GruntParser extends PigScriptParser {
private static final Log log = LogFactory.getLog(GruntParser.class);
@@ -854,12 +855,13 @@ public class GruntParser extends PigScri
System.out.println(" fs <fs arguments> - Equivalent to Hadoop dfs command: http://hadoop.apache.org/common/docs/current/hdfs_shell.html");
System.out.println("Diagnostic commands:");
System.out.println(" describe <alias>[::<alias] - Show the schema for the alias. Inner aliases can be described as A::B.");
- System.out.println(" explain [-script <pigscript>] [-out <path>] [-brief] [-dot] [-param <param_name>=<param_value>]");
+ System.out.println(" explain [-script <pigscript>] [-out <path>] [-brief] [-dot|-xml] [-param <param_name>=<param_value>]");
System.out.println(" [-param_file <file_name>] [<alias>] - Show the execution plan to compute the alias or for entire script.");
System.out.println(" -script - Explain the entire script.");
System.out.println(" -out - Store the output into directory rather than print to stdout.");
System.out.println(" -brief - Don't expand nested plans (presenting a smaller graph for overview).");
System.out.println(" -dot - Generate the output in .dot format. Default is text format.");
+ System.out.println(" -xml - Generate the output in .xml format. Default is text format.");
System.out.println(" -param <param_name - See parameter substitution for details.");
System.out.println(" -param_file <file_name> - See parameter substitution for details.");
System.out.println(" alias - Alias to explain.");
@@ -1012,19 +1014,41 @@ public class GruntParser extends PigScri
@Override
protected void processRemove(String path, String options ) throws IOException
{
+ int MAX_MS_TO_WAIT_FOR_FILE_DELETION = 10 * 60 * 1000;
+ int MS_TO_SLEEP_WHILE_WAITING_FOR_FILE_DELETION = 250;
+
if(mExplain == null) { // process only if not in "explain" mode
+ Path filePath = new Path(path);
+ ElementDescriptor dfsPath = null;
+ FileSystem fs = filePath.getFileSystem(ConfigurationUtil.toConfiguration(mConf));
- ElementDescriptor dfsPath = mDfs.asElement(path);
executeBatch();
-
- if (!dfsPath.exists()) {
+ if (!fs.exists(filePath)) {
if (options == null || !options.equalsIgnoreCase("force")) {
throw new IOException("File or directory " + path + " does not exist.");
}
}
else {
-
- dfsPath.delete();
+ boolean deleteSuccess = fs.delete(filePath, true);
+ if (!deleteSuccess) {
+ log.warn("Unable to delete " + path);
+ return;
+ }
+ long startTime = System.currentTimeMillis();
+ long duration = 0;
+ while(fs.exists(filePath)) {
+ duration = System.currentTimeMillis() - startTime;
+ if (duration > MAX_MS_TO_WAIT_FOR_FILE_DELETION) {
+ throw new IOException("Timed out waiting to delete file: " + dfsPath);
+ } else {
+ try {
+ Thread.sleep(MS_TO_SLEEP_WHILE_WAITING_FOR_FILE_DELETION);
+ } catch (InterruptedException e) {
+ throw new IOException("Error waiting for file deletion", e);
+ }
+ }
+ }
+ log.info("Waited " + duration + "ms to delete file");
}
} else {
log.warn("'rm/rmf' statement is ignored while processing 'explain -script' or '-check'");
@@ -1209,7 +1233,7 @@ public class GruntParser extends PigScri
}
}
- private static class ExplainState {
+ protected static class ExplainState {
public long mTime;
public int mCount;
public String mAlias;
@@ -1243,5 +1267,10 @@ public class GruntParser extends PigScri
private int mNumSucceededJobs;
private FsShell shell;
private boolean mScriptIllustrate;
+
+ //For Testing Only
+ protected void setExplainState(ExplainState explainState) {
+ this.mExplain = explainState;
+ }
}
Modified: pig/branches/tez/src/org/apache/pig/tools/parameters/PreprocessorContext.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/tools/parameters/PreprocessorContext.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/tools/parameters/PreprocessorContext.java (original)
+++ pig/branches/tez/src/org/apache/pig/tools/parameters/PreprocessorContext.java Wed Sep 18 20:21:06 2013
@@ -35,6 +35,7 @@ import java.util.regex.Pattern;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.util.Shell;
public class PreprocessorContext {
@@ -75,7 +76,7 @@ public class PreprocessorContext {
* @param key - parameter name
* @param val - string containing command to be executed
*/
- public void processShellCmd(String key, String val) {
+ public void processShellCmd(String key, String val) throws ParameterSubstitutionException {
processShellCmd(key, val, true);
}
@@ -86,7 +87,7 @@ public class PreprocessorContext {
* @param key - parameter name
* @param val - value supplied for the key
*/
- public void processOrdLine(String key, String val) {
+ public void processOrdLine(String key, String val) throws ParameterSubstitutionException {
processOrdLine(key, val, true);
}
@@ -111,7 +112,7 @@ public class PreprocessorContext {
* @param key - parameter name
* @param val - string containing command to be executed
*/
- public void processShellCmd(String key, String val, Boolean overwrite) {
+ public void processShellCmd(String key, String val, Boolean overwrite) throws ParameterSubstitutionException {
if (param_val.containsKey(key)) {
if (param_source.get(key).equals(val) || !overwrite) {
@@ -137,7 +138,7 @@ public class PreprocessorContext {
* @param val - value supplied for the key
* @param overwrite - specifies whether the value should be replaced if it already exists
*/
- public void processOrdLine(String key, String val, Boolean overwrite) {
+ public void processOrdLine(String key, String val, Boolean overwrite) throws ParameterSubstitutionException {
if (param_val.containsKey(key)) {
if (param_source.get(key).equals(val) || !overwrite) {
@@ -149,7 +150,7 @@ public class PreprocessorContext {
param_source.put(key, val);
- String sub_val = substitute(val);
+ String sub_val = substitute(val, key);
param_val.put(key, sub_val);
}
@@ -166,12 +167,17 @@ public class PreprocessorContext {
log.info("Executing command : " + cmd);
// we can't use exec directly since it does not handle
// case like foo -c "bar bar" correctly. It splits on white spaces even in presents of quotes
- String[] cmdArgs = new String[3];
- cmdArgs[0] = "bash";
- cmdArgs[1] = "-c";
- StringBuffer sb = new StringBuffer("exec ");
- sb.append(cmd);
- cmdArgs[2] = sb.toString();
+ StringBuffer sb = new StringBuffer("");
+ String[] cmdArgs;
+ if (Shell.WINDOWS) {
+ cmd = cmd.replaceAll("/", "\\\\");
+ sb.append(cmd);
+ cmdArgs = new String[]{"cmd", "/c", sb.toString() };
+ } else {
+ sb.append("exec ");
+ sb.append(cmd);
+ cmdArgs = new String[]{"bash", "-c", sb.toString() };
+ }
p = Runtime.getRuntime().exec(cmdArgs);
@@ -258,10 +264,14 @@ public class PreprocessorContext {
private Pattern bracketIdPattern = Pattern.compile("\\$\\{([_]*[a-zA-Z][a-zA-Z_0-9]*)\\}");
private Pattern id_pattern = Pattern.compile("\\$([_]*[a-zA-Z][a-zA-Z_0-9]*)");
- public String substitute(String line) {
+ public String substitute(String line) throws ParameterSubstitutionException {
+ return substitute(line, null);
+ }
+ public String substitute(String line, String parentKey) throws ParameterSubstitutionException {
int index = line.indexOf('$');
- if (index == -1) return line;
+ if (index == -1)
+ return line;
String replaced_line = line;
@@ -274,7 +284,13 @@ public class PreprocessorContext {
if ( (bracketKeyMatcher.start() == 0) || (line.charAt( bracketKeyMatcher.start() - 1)) != '\\' ) {
key = bracketKeyMatcher.group(1);
if (!(param_val.containsKey(key))) {
- throw new RuntimeException("Undefined parameter : "+key);
+ String message;
+ if (parentKey == null) {
+ message = "Undefined parameter : " + key;
+ } else {
+ message = "Undefined parameter : " + key + " found when trying to find the value of " + parentKey + ".";
+ }
+ throw new ParameterSubstitutionException(message);
}
val = param_val.get(key);
if (val.contains("$")) {
@@ -295,7 +311,13 @@ public class PreprocessorContext {
if ( (keyMatcher.start() == 0) || (line.charAt( keyMatcher.start() - 1)) != '\\' ) {
key = keyMatcher.group(1);
if (!(param_val.containsKey(key))) {
- throw new RuntimeException("Undefined parameter : "+key);
+ String message;
+ if (parentKey == null) {
+ message = "Undefined parameter : " + key;
+ } else {
+ message = "Undefined parameter : " + key + " found when trying to find the value of " + parentKey + ".";
+ }
+ throw new ParameterSubstitutionException(message);
}
val = param_val.get(key);
if (val.contains("$")) {
Modified: pig/branches/tez/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj (original)
+++ pig/branches/tez/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj Wed Sep 18 20:21:06 2013
@@ -180,6 +180,7 @@ TOKEN: {<PARAM: "-param">}
TOKEN: {<PARAM_FILE: "-param_file">}
TOKEN: {<SCRIPT: "-script">}
TOKEN: {<DOT: "-dot">}
+TOKEN: {<XML: "-xml">}
TOKEN: {<OUT: "-out">}
TOKEN: {<BRIEF: "-brief">}
TOKEN: {<N: "-n">}
@@ -691,6 +692,9 @@ void Explain() throws IOException:
<DOT>
{format = "dot";}
|
+ <XML>
+ {format = "xml";}
+ |
<OUT>
t = GetPath()
{target = t.image;}
Modified: pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStats.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStats.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStats.java (original)
+++ pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStats.java Wed Sep 18 20:21:06 2013
@@ -60,6 +60,7 @@ public abstract class PigStats {
private String errorMessage;
private int errorCode = -1;
+ private Throwable errorThrowable = null;
public static PigStats get() {
if (tps.get() == null) {
@@ -98,6 +99,13 @@ public abstract class PigStats {
public int getErrorCode() {
return errorCode;
}
+
+ /**
+ * Returns the error code of {@link PigException}
+ */
+ public Throwable getErrorThrowable() {
+ return errorThrowable;
+ }
public abstract JobClient getJobClient();
@@ -205,7 +213,10 @@ public abstract class PigStats {
void setErrorCode(int errorCode) {
this.errorCode = errorCode;
}
-
+
+ void setErrorThrowable(Throwable t) {
+ this.errorThrowable = t;
+ }
/**
* This class prints a JobGraph
Modified: pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStatsUtil.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStatsUtil.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStatsUtil.java (original)
+++ pig/branches/tez/src/org/apache/pig/tools/pigstats/PigStatsUtil.java Wed Sep 18 20:21:06 2013
@@ -66,7 +66,11 @@ public class PigStatsUtil {
public static void setErrorCode(int code) {
PigStats.get().setErrorCode(code);
}
-
+
+ public static void setErrorThrowable(Throwable t) {
+ PigStats.get().setErrorThrowable(t);
+ }
+
private static Pattern pattern = Pattern.compile("tmp(-)?[\\d]{1,10}$");
public static boolean isTempFile(String fileName) {
Modified: pig/branches/tez/src/pig-default.properties
URL: http://svn.apache.org/viewvc/pig/branches/tez/src/pig-default.properties?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/src/pig-default.properties (original)
+++ pig/branches/tez/src/pig-default.properties Wed Sep 18 20:21:06 2013
@@ -53,3 +53,5 @@ pig.optimistic.files.concatenation=false
pig.disable.counter=false
pig.sql.type=hcat
+
+pig.output.committer.recovery.support=false
Propchange: pig/branches/tez/src/pig-default.properties
------------------------------------------------------------------------------
Merged /pig/trunk/src/pig-default.properties:r1519486-1524491
Modified: pig/branches/tez/test/org/apache/pig/TestMain.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/TestMain.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/TestMain.java (original)
+++ pig/branches/tez/test/org/apache/pig/TestMain.java Wed Sep 18 20:21:06 2013
@@ -18,14 +18,29 @@
package org.apache.pig;
-import static org.junit.Assert.*;
-
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
import java.util.Properties;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.parser.ParserException;
+import org.apache.pig.parser.SourceLocation;
import org.apache.pig.test.TestPigRunner.TestNotificationListener;
+import org.apache.pig.tools.parameters.ParameterSubstitutionException;
+import org.apache.pig.tools.pigstats.PigStats;
import org.junit.Test;
public class TestMain {
+ private Log log = LogFactory.getLog(TestMain.class);
@Test
public void testCustomListener() {
@@ -41,6 +56,75 @@ public class TestMain {
assertTrue(listener.hadArgs);
}
+
+ @Test
+ public void testRun_setsErrorThrowableOnPigStats() {
+ File outputFile = null;
+ try {
+ String filename = this.getClass().getSimpleName() + "_" + "testRun_setsErrorThrowableOnPigStats";
+ outputFile = File.createTempFile(filename, ".out");
+ outputFile.delete();
+
+ File scriptFile = File.createTempFile(filename, ".pig");
+ BufferedWriter bw = new BufferedWriter(new FileWriter(scriptFile));
+ bw.write("a = load 'test/org/apache/pig/test/data/passwd';\n");
+ bw.write("b = group a by $0\n");
+ bw.write("c = foreach b generate group, COUNT(a) as cnt;\n");
+ bw.write("store c into 'out'\n");
+ bw.close();
+
+ Main.run(new String[]{"-x", "local", scriptFile.getAbsolutePath()}, null);
+ PigStats stats = PigStats.get();
+
+ Throwable t = stats.getErrorThrowable();
+ assertTrue(t instanceof FrontendException);
+
+ FrontendException fe = (FrontendException) t;
+ SourceLocation sl = fe.getSourceLocation();
+ assertEquals(2, sl.line());
+ assertEquals(15, sl.offset());
+
+ Throwable cause = fe.getCause();
+ assertTrue(cause instanceof ParserException);
+
+ } catch (Exception e) {
+ log.error("Encountered exception", e);
+ fail("Encountered Exception");
+ }
+ }
+
+ @Test
+ public void testRun_setsErrorThrowableForParamSubstitution() {
+ File outputFile = null;
+ try {
+ String filename = this.getClass().getSimpleName() + "_" + "testRun_setsErrorThrowableForParamSubstitution";
+ outputFile = File.createTempFile(filename, ".out");
+ outputFile.delete();
+
+ File scriptFile = File.createTempFile(filename, ".pig");
+ BufferedWriter bw = new BufferedWriter(new FileWriter(scriptFile));
+ bw.write("a = load '$NOEXIST';\n");
+ bw.write("b = group a by $0\n");
+ bw.write("c = foreach b generate group, COUNT(a) as cnt;\n");
+ bw.write("store c into 'out'\n");
+ bw.close();
+
+ Main.run(new String[]{"-x", "local", scriptFile.getAbsolutePath()}, null);
+ PigStats stats = PigStats.get();
+
+ Throwable t = stats.getErrorThrowable();
+ assertTrue(t instanceof IOException);
+
+ Throwable cause = t.getCause();
+ assertTrue(cause instanceof ParameterSubstitutionException);
+
+ ParameterSubstitutionException pse = (ParameterSubstitutionException) cause;
+ assertTrue(pse.getMessage().contains("NOEXIST"));
+ } catch (Exception e) {
+ log.error("Encountered exception", e);
+ fail("Encountered Exception");
+ }
+ }
public static class TestNotificationListener2 extends TestNotificationListener {
protected boolean hadArgs = false;
Modified: pig/branches/tez/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java (original)
+++ pig/branches/tez/test/org/apache/pig/newplan/logical/relational/TestLocationInPhysicalPlan.java Wed Sep 18 20:21:06 2013
@@ -49,10 +49,10 @@ public class TestLocationInPhysicalPlan
PigServer pigServer = new PigServer(ExecType.LOCAL);
pigServer.setBatchOn();
pigServer.registerQuery(
- "A = LOAD '" + input.getAbsolutePath() + "' using PigStorage();\n"
+ "A = LOAD '" + Util.encodeEscape(input.getAbsolutePath()) + "' using PigStorage();\n"
+ "B = GROUP A BY $0;\n"
+ "A = FOREACH B GENERATE COUNT(A);\n"
- + "STORE A INTO '" + output.getAbsolutePath() + "';");
+ + "STORE A INTO '" + Util.encodeEscape(output.getAbsolutePath()) + "';");
ExecJob job = pigServer.executeBatch().get(0);
List<OriginalLocation> originalLocations = job.getPOStore().getOriginalLocations();
Assert.assertEquals(1, originalLocations.size());
Modified: pig/branches/tez/test/org/apache/pig/test/TestBZip.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestBZip.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestBZip.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestBZip.java Wed Sep 18 20:21:06 2013
@@ -194,6 +194,7 @@ public class TestBZip {
// bzip compressed input
File in = File.createTempFile("junit", ".bz2");
String compressedInputFileName = in.getAbsolutePath();
+ String clusterCompressedFilePath = Util.removeColon(compressedInputFileName);
in.deleteOnExit();
try {
@@ -208,14 +209,14 @@ public class TestBZip {
cos.close();
Util.copyFromLocalToCluster(cluster, compressedInputFileName,
- compressedInputFileName);
+ clusterCompressedFilePath);
// pig script to read compressed input
PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster
.getProperties());
// pig script to read compressed input
- String script ="a = load '" + compressedInputFileName +"';";
+ String script ="a = load '" + Util.encodeEscape(clusterCompressedFilePath) +"';";
pig.registerQuery(script);
pig.registerQuery("store a into 'intermediate.bz';");
Modified: pig/branches/tez/test/org/apache/pig/test/TestDefaultDateTimeZone.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestDefaultDateTimeZone.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestDefaultDateTimeZone.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestDefaultDateTimeZone.java Wed Sep 18 20:21:06 2013
@@ -70,7 +70,7 @@ public class TestDefaultDateTimeZone ext
config.setProperty("pig.datetime.default.tz", "+08:00");
PigServer pig = new PigServer(ExecType.LOCAL, config);
pig.registerQuery("a = load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' as (test:datetime);");
pig.registerQuery("b = filter a by test < ToDate('1970-01-04T00:00:00.000');");
Iterator<Tuple> actualItr = pig.openIterator("b");
Modified: pig/branches/tez/test/org/apache/pig/test/TestEvalPipelineLocal.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestEvalPipelineLocal.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestEvalPipelineLocal.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestEvalPipelineLocal.java Wed Sep 18 20:21:06 2013
@@ -1131,7 +1131,7 @@ public class TestEvalPipelineLocal {
// See PIG-2970
public void testDescribeDanglingBranch() throws Throwable {
File f1 = createFile(new String[]{"NYSE\tIBM", "NASDAQ\tYHOO", "NASDAQ\tMSFT"});
- pigServer.registerQuery("daily = load '" + Util.generateURI(f1.toString(), pigServer.getPigContext())
+ pigServer.registerQuery("daily = load '" + Util.encodeEscape(Util.generateURI(f1.toString(), pigServer.getPigContext()))
+"' as (exchange, symbol);");
pigServer.registerQuery("grpd = group daily by exchange;");
pigServer.registerQuery("unique = foreach grpd { sym = daily.symbol; uniq_sym = distinct sym; uniq_sym = distinct sym; generate group, daily;};");
Modified: pig/branches/tez/test/org/apache/pig/test/TestFilterOpNumeric.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestFilterOpNumeric.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestFilterOpNumeric.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestFilterOpNumeric.java Wed Sep 18 20:21:06 2013
@@ -60,7 +60,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using "+PigStorage.class.getName() +"(':');");
String query = "A = filter A by ($0 == $1 and $0 <= $1);";
log.info(query);
@@ -92,7 +92,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 != $1;";
log.info(query);
@@ -120,7 +120,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext()) + "' using "
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext())) + "' using "
+ PigStorage.class.getName() + "(':') as (f1: double, f2:double);");
String query = "A = filter A by ($0 > $1 and $0 >= $1);";
@@ -145,7 +145,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext()) + "';");
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext())) + "';");
String query = "A = foreach A generate ($1 >= "+ LOOP_COUNT+"-10?'1':'0');";
log.info(query);
pig.registerQuery(query);
@@ -174,7 +174,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext()) + "';");
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext())) + "';");
String query = "A = foreach A generate (($0 < 10 or $0 < 9)?(($1 >= 5 and $1 >= 4) ? 2: 1) : 0);";
log.info(query);
pig.registerQuery(query);
@@ -204,7 +204,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':') as (a: double, b:double);");
String query = "A = filter A by ($0 <= $1 and $0 < $1);";
@@ -236,7 +236,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by ($0 > $1 or $0 >= $1);";
@@ -267,7 +267,7 @@ public class TestFilterOpNumeric {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':') as (a: double, b:double);");
String query = "A = filter A by ($0 <= $1 or $0 < $1);";
Modified: pig/branches/tez/test/org/apache/pig/test/TestFilterOpString.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestFilterOpString.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestFilterOpString.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestFilterOpString.java Wed Sep 18 20:21:06 2013
@@ -68,7 +68,7 @@ public class TestFilterOpString {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 eq $1;";
@@ -107,7 +107,7 @@ public class TestFilterOpString {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 neq $1;";
@@ -146,7 +146,7 @@ public class TestFilterOpString {
}
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 gt $1;";
@@ -190,7 +190,7 @@ public class TestFilterOpString {
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 gte $1;";
@@ -229,7 +229,7 @@ public class TestFilterOpString {
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 lt $1;";
@@ -271,7 +271,7 @@ public class TestFilterOpString {
ps.close();
pig.registerQuery("A=load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName() + "(':');");
String query = "A = filter A by $0 lte $1;";
Modified: pig/branches/tez/test/org/apache/pig/test/TestFinish.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestFinish.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestFinish.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestFinish.java Wed Sep 18 20:21:06 2013
@@ -141,7 +141,7 @@ public class TestFinish {
String expectedFileName = "testFinishInMapMR-finish.txt";
pigServer.registerQuery("define MYUDF " + MyEvalFunction.class.getName() + "('MAPREDUCE','"
+ expectedFileName + "');");
- pigServer.registerQuery("a = load '" + inputFileName + "' using "
+ pigServer.registerQuery("a = load '" + Util.encodeEscape(inputFileName) + "' using "
+ PigStorage.class.getName() + "(':');");
pigServer.registerQuery("b = foreach a generate MYUDF" + "(*);");
Iterator<Tuple> iter = pigServer.openIterator("b");
@@ -160,7 +160,7 @@ public class TestFinish {
String expectedFileName = "testFinishInReduceMR-finish.txt";
pigServer.registerQuery("define MYUDF " + MyEvalFunction.class.getName() + "('MAPREDUCE','"
+ expectedFileName + "');");
- pigServer.registerQuery("a = load '" + inputFileName + "' using "
+ pigServer.registerQuery("a = load '" + Util.encodeEscape(inputFileName) + "' using "
+ PigStorage.class.getName() + "(':');");
pigServer.registerQuery("a1 = group a by $1;");
pigServer.registerQuery("b = foreach a1 generate MYUDF" + "(*);");
@@ -179,7 +179,7 @@ public class TestFinish {
String expectedFileName = "testFinishInMapLoc-finish.txt";
pigServer.registerQuery("define MYUDF " + MyEvalFunction.class.getName() + "('LOCAL','"
+ expectedFileName + "');");
- pigServer.registerQuery("a = load '" + inputFileName + "' using "
+ pigServer.registerQuery("a = load '" + Util.encodeEscape(inputFileName) + "' using "
+ PigStorage.class.getName() + "(':');");
pigServer.registerQuery("b = foreach a generate MYUDF" + "(*);");
pigServer.openIterator("b");
@@ -193,7 +193,7 @@ public class TestFinish {
String expectedFileName = "testFinishInReduceLoc-finish.txt";
pigServer.registerQuery("define MYUDF " + MyEvalFunction.class.getName() + "('LOCAL','"
+ expectedFileName + "');");
- pigServer.registerQuery("a = load '" + inputFileName + "' using "
+ pigServer.registerQuery("a = load '" + Util.encodeEscape(inputFileName) + "' using "
+ PigStorage.class.getName() + "(':');");
pigServer.registerQuery("a1 = group a by $1;");
pigServer.registerQuery("b = foreach a1 generate MYUDF" + "(*);");
Modified: pig/branches/tez/test/org/apache/pig/test/TestForEachNestedPlanLocal.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestForEachNestedPlanLocal.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestForEachNestedPlanLocal.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestForEachNestedPlanLocal.java Wed Sep 18 20:21:06 2013
@@ -103,10 +103,10 @@ public class TestForEachNestedPlanLocal
"({('user3','singapore','user3','usa','10'),('user3','singapore','user3','singapore','20')})",
"({})" });
pig.registerQuery("user = load '"
- + Util.generateURI(tmpFiles[0].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[0].toString(), pig.getPigContext()))
+ "' as (uid, region);");
pig.registerQuery("session = load '"
- + Util.generateURI(tmpFiles[1].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[1].toString(), pig.getPigContext()))
+ "' as (uid, region, duration);");
pig.registerQuery("C = cogroup user by uid, session by uid;");
pig.registerQuery("D = foreach C {"
@@ -131,10 +131,10 @@ public class TestForEachNestedPlanLocal
"({('user3','singapore','user3','singapore','20')})",
"({})" });
pig.registerQuery("user = load '"
- + Util.generateURI(tmpFiles[0].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[0].toString(), pig.getPigContext()))
+ "' as (uid, region);");
pig.registerQuery("session = load '"
- + Util.generateURI(tmpFiles[1].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[1].toString(), pig.getPigContext()))
+ "' as (uid, region, duration);");
pig.registerQuery("C = cogroup user by uid, session by uid;");
pig.registerQuery("D = foreach C {"
@@ -161,13 +161,13 @@ public class TestForEachNestedPlanLocal
"({('user3','singapore','user3','usa','10','user3','user','female'),('user3','singapore','user3','singapore','20','user3','user','female')})",
"({})" });
pig.registerQuery("user = load '"
- + Util.generateURI(tmpFiles[0].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[0].toString(), pig.getPigContext()))
+ "' as (uid, region);");
pig.registerQuery("session = load '"
- + Util.generateURI(tmpFiles[1].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[1].toString(), pig.getPigContext()))
+ "' as (uid, region, duration);");
pig.registerQuery("profile = load '"
- + Util.generateURI(tmpFiles[2].toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFiles[2].toString(), pig.getPigContext()))
+ "' as (uid, role, gender);");
pig.registerQuery("C = cogroup user by uid, session by uid, profile by uid;");
pig.registerQuery("D = foreach C {"
Modified: pig/branches/tez/test/org/apache/pig/test/TestGrunt.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestGrunt.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestGrunt.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestGrunt.java Wed Sep 18 20:21:06 2013
@@ -77,7 +77,7 @@ public class TestGrunt {
PigServer server = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
PigContext context = server.getPigContext();
- String strCmd = "copyFromLocal /bin/sh sh_copy ;";
+ String strCmd = "copyFromLocal README.txt sh_copy ;";
ByteArrayInputStream cmd = new ByteArrayInputStream(strCmd.getBytes());
InputStreamReader reader = new InputStreamReader(cmd);
@@ -1116,11 +1116,11 @@ public class TestGrunt {
public void testShellCommandOrder() throws Throwable {
PigServer server = new PigServer(ExecType.LOCAL, new Properties());
- String strRemoveRecurse = "rm -rf";
+ String strRemove = "rm";
if (Util.WINDOWS)
{
- strRemoveRecurse = "rd /S";
+ strRemove = "del";
}
File inputFile = File.createTempFile("testInputFile", ".txt");
@@ -1134,7 +1134,7 @@ public class TestGrunt {
PrintWriter pwScript = new PrintWriter(new FileWriter(inputScript));
pwScript.println("a = load '" + Util.encodeEscape(inputFile.getAbsolutePath()) + "';");
pwScript.println("store a into '" + Util.encodeEscape(outputFile.getAbsolutePath()) + "';");
- pwScript.println("sh " + strRemoveRecurse + " " + Util.encodeEscape(inputFile.getAbsolutePath()));
+ pwScript.println("sh " + strRemove + " " + Util.encodeEscape(inputFile.getAbsolutePath()));
pwScript.close();
InputStream inputStream = new FileInputStream(inputScript.getAbsoluteFile());
Modified: pig/branches/tez/test/org/apache/pig/test/TestHBaseStorage.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestHBaseStorage.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestHBaseStorage.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestHBaseStorage.java Wed Sep 18 20:21:06 2013
@@ -38,7 +38,6 @@ import org.apache.hadoop.hbase.util.Byte
import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.backend.executionengine.ExecException;
-import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
import org.apache.pig.backend.hadoop.hbase.HBaseStorage;
import org.apache.pig.data.DataByteArray;
import org.apache.pig.data.Tuple;
@@ -60,8 +59,6 @@ public class TestHBaseStorage {
private static MiniCluster cluster;
private static PigServer pig;
- final static int NUM_REGIONSERVERS = 1;
-
enum DataFormat {
HBaseBinary, UTF8PlainText,
}
Modified: pig/branches/tez/test/org/apache/pig/test/TestInfixArithmetic.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestInfixArithmetic.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestInfixArithmetic.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestInfixArithmetic.java Wed Sep 18 20:21:06 2013
@@ -60,7 +60,7 @@ public class TestInfixArithmetic {
PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
generateInput(ps, nullFlags[i]);
String query = "A = foreach (load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName()
+ "(':')) generate $0, $0 + $1, $1;";
log.info(query);
@@ -89,7 +89,7 @@ public class TestInfixArithmetic {
PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
generateInput(ps, nullFlags[i]);
String query = "A = foreach (load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName()
+ "(':')) generate $0, $0 - $1, $1 ;";
log.info(query);
@@ -118,7 +118,7 @@ public class TestInfixArithmetic {
PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
generateInput(ps, nullFlags[i]);
String query = "A = foreach (load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName()
+ "(':')) generate $0, $0 * $1, $1 ;";
log.info(query);
@@ -147,7 +147,7 @@ public class TestInfixArithmetic {
PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
generateInput(ps, nullFlags[i]);
String query = "A = foreach (load '"
- + Util.generateURI(tmpFile.toString(), pig.getPigContext())
+ + Util.encodeEscape(Util.generateURI(tmpFile.toString(), pig.getPigContext()))
+ "' using " + PigStorage.class.getName()
+ "(':')) generate $0, $0 / $1, $1 ;";
log.info(query);
Modified: pig/branches/tez/test/org/apache/pig/test/TestInputOutputMiniClusterFileValidator.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestInputOutputMiniClusterFileValidator.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestInputOutputMiniClusterFileValidator.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestInputOutputMiniClusterFileValidator.java Wed Sep 18 20:21:06 2013
@@ -219,10 +219,8 @@ public class TestInputOutputMiniClusterF
ctx.getLfs().asElement(fp1.getAbsolutePath());
String path = fp1.getAbsolutePath();
- if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS"))
- path = FileLocalizer.parseCygPath(path, FileLocalizer.STYLE_UNIX);
- ElementDescriptor distribElem = ctx.getDfs().asElement(path) ;
+ ElementDescriptor distribElem = ctx.getDfs().asElement(Util.removeColon(path)) ;
localElem.copy(distribElem, null, false);
@@ -234,10 +232,8 @@ public class TestInputOutputMiniClusterF
File fp1 = generateTempFile() ;
String path = fp1.getAbsolutePath();
- if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS"))
- path = FileLocalizer.parseCygPath(path, FileLocalizer.STYLE_UNIX);
- ElementDescriptor distribElem = ctx.getDfs().asElement(path) ;
+ ElementDescriptor distribElem = ctx.getDfs().asElement(Util.removeColon(path)) ;
if (distribElem.exists()) {
distribElem.delete() ;
Modified: pig/branches/tez/test/org/apache/pig/test/TestJsonLoaderStorage.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestJsonLoaderStorage.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestJsonLoaderStorage.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestJsonLoaderStorage.java Wed Sep 18 20:21:06 2013
@@ -1,126 +1,291 @@
-/*
- * 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.test;
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecJob.JOB_STATUS;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.test.Util;
+
import java.io.File;
import java.io.FileWriter;
-import java.io.PrintWriter;
+import java.io.FileReader;
+import java.io.BufferedReader;
+import java.io.IOException;
+
import java.util.Iterator;
+import java.util.Map;
-import junit.framework.Assert;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+
+import org.joda.time.DateTime;
-import org.apache.pig.ExecType;
-import org.apache.pig.PigServer;
-import org.apache.pig.data.Tuple;
-import org.apache.pig.test.utils.TestHelper;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
import org.junit.Test;
+import static junit.framework.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
public class TestJsonLoaderStorage {
- private static PigServer pigServer;
- File jsonFile;
- @BeforeClass
- public static void setUp() throws Exception{
- removeOutput();
- pigServer = new PigServer(ExecType.LOCAL);
+ private static final String schema =
+ "a:boolean," +
+ "b:int," +
+ "c:long," +
+ "d:float," +
+ "e:double," +
+ "f:datetime," +
+ "g:bytearray," +
+ "h:chararray," +
+ "i:biginteger," +
+ "j:bigdecimal," +
+ "k:map[chararray]," +
+ "l:tuple(a:int)," +
+ "m:bag{a:tuple(a:int)}";
+
+ private static final String rawInput =
+ "true\t" +
+ "123\t" +
+ "456\t" +
+ "7.89\t" +
+ "0.12\t" +
+ "2013-01-02T03:04:05.123+03:00\t" +
+ "abc\t" +
+ "def\t" +
+ "123456789\t" +
+ "1234.6789\t" +
+ "[a#ghi]\t" +
+ "(123)\t" +
+ "{(123),(456),(789)}";
+
+ private static final String nullInput =
+ "\t\t\t\t\t\t\t\t\t\t\t\t";
+
+ private static final String json =
+ "{" +
+ "\"a\":true," +
+ "\"b\":123," +
+ "\"c\":456," +
+ "\"d\":7.89," +
+ "\"e\":0.12," +
+ "\"f\":\"2013-01-02T03:04:05.123+03:00\"," +
+ "\"g\":\"abc\"," +
+ "\"h\":\"def\"," +
+ "\"i\":123456789," +
+ "\"j\":1234.6789," +
+ "\"k\":{\"a\":\"ghi\"}," +
+ "\"l\":{\"a\":123}," +
+ "\"m\":[{\"a\":123},{\"a\":456},{\"a\":789}]" +
+ "}";
+
+ private static final String nullJson =
+ "{" +
+ "\"a\":null," +
+ "\"b\":null," +
+ "\"c\":null," +
+ "\"d\":null," +
+ "\"e\":null," +
+ "\"f\":null," +
+ "\"g\":null," +
+ "\"h\":null," +
+ "\"i\":null," +
+ "\"j\":null," +
+ "\"k\":null," +
+ "\"l\":null," +
+ "\"m\":null" +
+ "}";
+
+ private Iterator<Tuple> loadJson(String input) throws IOException {
+ File tempFile = File.createTempFile("json", null);
+ tempFile.deleteOnExit();
+
+ FileWriter writer = new FileWriter(tempFile);
+ writer.write(input);
+ writer.close();
+
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ pigServer.registerQuery("data = load '" + tempFile.getAbsolutePath()
+ + "' using JsonLoader('" + schema + "');");
+
+ return pigServer.openIterator("data");
+ }
+
+ private BufferedReader storeJson(String input) throws IOException {
+ File tempJsonFile = File.createTempFile("json", "");
+ tempJsonFile.delete();
+
+ File tempInputFile = File.createTempFile("input", null);
+ tempInputFile.deleteOnExit();
+
+ FileWriter w = new FileWriter(tempInputFile);
+ w.write(input);
+ w.close();
+
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ pigServer.registerQuery("data = load '" + tempInputFile.getAbsolutePath()
+ + "' as (" + schema + ");");
+ pigServer.registerQuery("store data into '" + tempJsonFile.getAbsolutePath()
+ + "' using JsonStorage();");
+
+ tempJsonFile.deleteOnExit();
+
+ FileReader r = new FileReader(tempJsonFile.getAbsolutePath() + "/part-m-00000");
+ BufferedReader br = new BufferedReader(r);
+
+ return br;
+ }
+
+ @Test
+ public void testJsonLoader() throws IOException {
+ Iterator<Tuple> tuples = loadJson(json);
+
+ int count = 0;
+ while(tuples.hasNext()) {
+ Tuple tuple = tuples.next();
+ assertEquals(Boolean.class, tuple.get(0).getClass());
+ assertEquals(Integer.class, tuple.get(1).getClass());
+ assertEquals(Long.class, tuple.get(2).getClass());
+ assertEquals(Float.class, tuple.get(3).getClass());
+ assertEquals(Double.class, tuple.get(4).getClass());
+
+ assertEquals(DateTime.class, tuple.get(5).getClass());
+ assertEquals("2013-01-02T03:04:05.123+03:00", tuple.get(5).toString());
+
+ assertEquals(DataByteArray.class, tuple.get(6).getClass());
+ assertEquals(String.class, tuple.get(7).getClass());
+ assertEquals(BigInteger.class, tuple.get(8).getClass());
+ assertEquals(BigDecimal.class, tuple.get(9).getClass());
+
+ assertTrue(tuple.get(10) instanceof Map);
+ assertEquals(String.class, ((Map)tuple.get(10)).get("a").getClass());
+
+ assertTrue(tuple.get(11) instanceof Tuple);
+ assertEquals(Integer.class, ((Tuple)tuple.get(11)).get(0).getClass());
+
+ assertTrue(tuple.get(12) instanceof DataBag);
+
+ DataBag bag = (DataBag)tuple.get(12);
+ assertEquals(3, bag.size());
+
+ Iterator<Tuple> bagTuples = bag.iterator();
+ while(bagTuples.hasNext()) {
+ assertEquals(Integer.class, bagTuples.next().get(0).getClass());
+ }
+
+ count++;
}
-
- private static void removeOutput() {
- File outputDir = new File("jsonStorage1.json");
- if (outputDir.exists()) {
- for (File c : outputDir.listFiles())
- c.delete();
- outputDir.delete();
- }
+
+ assertEquals(1, count);
+ }
+
+ @Test
+ public void testJsonLoaderNull() throws IOException {
+ Iterator<Tuple> tuples = loadJson(nullJson);
+
+ int count = 0;
+ while(tuples.hasNext()) {
+ Tuple tuple = tuples.next();
+
+ assertEquals(null, tuple.get(0));
+ assertEquals(null, tuple.get(1));
+ assertEquals(null, tuple.get(2));
+ assertEquals(null, tuple.get(3));
+ assertEquals(null, tuple.get(4));
+ assertEquals(null, tuple.get(5));
+ assertEquals(null, tuple.get(6));
+ assertEquals(null, tuple.get(7));
+ assertEquals(null, tuple.get(8));
+ assertEquals(null, tuple.get(9));
+ assertEquals(null, tuple.get(10));
+ assertEquals(null, tuple.get(11));
+ assertEquals(null, tuple.get(12));
+
+ count++;
}
-
- @Test
- public void testJsonLoaderStorage1() throws Exception{
- removeOutput();
-
- pigServer.registerScript("test/org/apache/pig/test/data/jsonStorage1.pig");
-
- File resultFile = new File("jsonStorage1.json/part-m-00000");
-
- String result = Util.readFile(resultFile);
- String expected = Util.readFile(new File("test/org/apache/pig/test/data/jsonStorage1.result"));
- Assert.assertEquals(TestHelper.sortStringList(expected,"{","}",","), TestHelper.sortStringList(result,"{","}",","));
-
- File schemaFile = new File("jsonStorage1.json/.pig_schema");
- result = Util.readFile(schemaFile);
- expected = Util.readFile(new File("test/org/apache/pig/test/data/jsonStorage1.schema"));
- Assert.assertEquals(expected, result);
-
- File tmpFile = File.createTempFile("tmp", null);
- tmpFile.delete();
-
- pigServer.registerQuery("a = load 'jsonStorage1.json' using JsonLoader();");
- pigServer.store("a", tmpFile.getCanonicalPath());
-
- result = Util.readFile(new File(tmpFile.getCanonicalPath()+"/part-m-00000"));
- expected = Util.readFile(new File("test/org/apache/pig/test/data/jsonStorage1.txt"));
- Assert.assertEquals(TestHelper.sortStringList(expected,"[","]",","), TestHelper.sortStringList(result,"[","]",","));
-
-
- tmpFile = File.createTempFile("tmp", null);
- tmpFile.delete();
-
- pigServer.registerQuery("a = load 'jsonStorage1.json' using" +
- " JsonLoader('a0:int,a1:{(a10:int,a11:chararray)},a2:(a20:double,a21:bytearray),a3:[chararray]');");
- pigServer.store("a", tmpFile.getCanonicalPath());
-
- result = Util.readFile(new File(tmpFile.getCanonicalPath()+"/part-m-00000"));
- expected = Util.readFile(new File("test/org/apache/pig/test/data/jsonStorage1.txt"));
- Assert.assertEquals(TestHelper.sortStringList(expected,"[","]",","), TestHelper.sortStringList(result,"[","]",","));
+
+ assertEquals(1, count);
+ }
+
+ @Test
+ public void testJsonStorage() throws IOException {
+ BufferedReader br = storeJson(rawInput);
+ String data = br.readLine();
+
+ assertEquals(json, data);
+
+ String line = data;
+ int count = 0;
+ while (line != null) {
+ line = br.readLine();
+ count++;
}
-
- @Test
- public void testJsonLoaderStorage2() throws Exception{
-
- File inputFile = File.createTempFile("tmp", null);
- PrintWriter pw = new PrintWriter(new FileWriter(inputFile));
- pw.println("\t\t\t");
- pw.close();
-
- File interFile = File.createTempFile("tmp", null);
- interFile.delete();
-
- pigServer.registerQuery("a = load '" + inputFile.getCanonicalPath() + "' as (a0:int, a1:chararray, a2, a3:(a30:int));");
- pigServer.store("a", interFile.getCanonicalPath(), "JsonStorage");
-
- pigServer.registerQuery("b = load '" + interFile.getCanonicalPath() + "' using JsonLoader();");
- Iterator<Tuple> iter = pigServer.openIterator("b");
-
- Tuple t = iter.next();
-
- Assert.assertTrue(t.size()==4);
- Assert.assertTrue(t.get(0)==null);
- Assert.assertTrue(t.get(1)==null);
- Assert.assertTrue(t.get(2)==null);
- Assert.assertTrue(t.get(3)==null);
-
- Assert.assertFalse(iter.hasNext());
+ assertEquals(1, count);
+
+ br.close();
+ }
+
+ @Test
+ public void testJsonStorageNull() throws IOException {
+ BufferedReader br = storeJson(nullInput);
+ String data = br.readLine();
+
+ assertEquals(nullJson, data);
+
+ String line = data;
+ int count = 0;
+ while (line != null) {
+ line = br.readLine();
+ count++;
}
+ assertEquals(1, count);
+
+ br.close();
+ }
+
+ @Test
+ public void testJsonLoaderStorage() throws IOException {
+ File tempJsonFile = File.createTempFile("json", "");
+ tempJsonFile.delete();
+
+ File tempJson2File = File.createTempFile("json2", "");
+ tempJson2File.delete();
+
+ File tempInputFile = File.createTempFile("input", null);
+ tempInputFile.deleteOnExit();
- @AfterClass
- public static void tearDown() {
- removeOutput();
+ FileWriter w = new FileWriter(tempInputFile);
+ w.write(rawInput);
+ w.close();
+
+ PigServer pigServer = new PigServer(ExecType.LOCAL);
+ pigServer.registerQuery("data = load '" + tempInputFile.getAbsolutePath()
+ + "' as (" + schema + ");");
+ pigServer.registerQuery("store data into '" + tempJsonFile.getAbsolutePath()
+ + "' using JsonStorage();");
+ pigServer.registerQuery("json = load '" + tempJsonFile.getAbsolutePath()
+ + "' using JsonLoader('" + schema + "');");
+ pigServer.registerQuery("store json into '" + tempJson2File.getAbsolutePath()
+ + "' using JsonStorage();");
+
+ tempJsonFile.deleteOnExit();
+ tempJson2File.deleteOnExit();
+
+ FileReader r = new FileReader(tempJson2File.getAbsolutePath() + "/part-m-00000");
+
+ BufferedReader br = new BufferedReader(r);
+ String data = br.readLine();
+
+ assertEquals(json, data);
+
+ String line = data;
+ int count = 0;
+ while (line != null) {
+ line = br.readLine();
+ count++;
}
+ assertEquals(1, count);
+
+ br.close();
+ }
}
Modified: pig/branches/tez/test/org/apache/pig/test/TestMacroExpansion.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestMacroExpansion.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestMacroExpansion.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestMacroExpansion.java Wed Sep 18 20:21:06 2013
@@ -334,7 +334,7 @@ public class TestMacroExpansion {
@Test
public void defineTest2() throws Exception {
String macro = "define group_and_count (A) returns B {\n" +
- " DEFINE CMD `stream.pl data.gz` SHIP('"+Util.encodeEscape(command.toString())+"') CACHE('"+Util.encodeEscape(command.toString())+"');\n" +
+ " DEFINE CMD `perl stream.pl data.gz` SHIP('"+Util.encodeEscape(command.toString())+"') CACHE('"+Util.encodeEscape(command.toString())+"');\n" +
" $B = STREAM $A THROUGH CMD;\n" +
"};\n";
Modified: pig/branches/tez/test/org/apache/pig/test/TestPONegative.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestPONegative.java?rev=1524555&r1=1524554&r2=1524555&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestPONegative.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestPONegative.java Wed Sep 18 20:21:06 2013
@@ -296,7 +296,7 @@ public class TestPONegative {
PigServer pig = new PigServer(ExecType.LOCAL, new Properties());
File f = Util.createInputFile("tmp", "", new String[] {"a", "b", "c"});
pig.registerQuery("a = load '"
- + Util.generateURI(f.toString(), pig.getPigContext()) + "';");
+ + Util.encodeEscape(Util.generateURI(f.toString(), pig.getPigContext())) + "';");
// -1 is modeled as POnegative with Constant(1)
pig.registerQuery("b = foreach a generate SIZE(-1);");
Iterator<Tuple> it = pig.openIterator("b");