You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2014/10/01 08:43:57 UTC
svn commit: r1628614 - in /pig/branches/branch-0.14: ./
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/
src/org/apache/pig/backend/hadoop/executionengine...
Author: daijy
Date: Wed Oct 1 06:43:56 2014
New Revision: 1628614
URL: http://svn.apache.org/r1628614
Log:
PIG-4175: PIG CROSS operation follow by STORE produces non-deterministic results each run
Modified:
pig/branches/branch-0.14/CHANGES.txt
pig/branches/branch-0.14/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java
pig/branches/branch-0.14/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java
pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java
pig/branches/branch-0.14/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
pig/branches/branch-0.14/test/org/apache/pig/test/TestEvalPipeline2.java
Modified: pig/branches/branch-0.14/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/CHANGES.txt?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/CHANGES.txt (original)
+++ pig/branches/branch-0.14/CHANGES.txt Wed Oct 1 06:43:56 2014
@@ -84,6 +84,8 @@ OPTIMIZATIONS
BUG FIXES
+PIG-4175: PIG CROSS operation follow by STORE produces non-deterministic results each run (daijy)
+
PIG-4202: Reset UDFContext state before OutputCommitter invocations in Tez (rohini)
PIG-4205: e2e test property-check does not check all prerequisites (kellyzly via daijy)
Modified: pig/branches/branch-0.14/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java (original)
+++ pig/branches/branch-0.14/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java Wed Oct 1 06:43:56 2014
@@ -103,6 +103,7 @@ public class CSVExcelStorage extends Pig
public static enum Headers { DEFAULT, READ_INPUT_HEADER, SKIP_INPUT_HEADER, WRITE_OUTPUT_HEADER, SKIP_OUTPUT_HEADER }
protected final static byte LINEFEED = '\n';
+ protected final static byte CARRIAGE_RETURN = '\r';
protected final static byte DOUBLE_QUOTE = '"';
protected final static byte RECORD_DEL = LINEFEED;
@@ -293,6 +294,7 @@ public class CSVExcelStorage extends Pig
ArrayList<Object> mProtoTuple = new ArrayList<Object>();
int embeddedNewlineIndex = -1;
+ int embeddedCarriageReturn = -1;
String fieldStr = null;
// For good debug messages:
int fieldCounter = -1;
@@ -315,13 +317,15 @@ public class CSVExcelStorage extends Pig
// If any field delimiters are in the field, or if we did replace
// any double quotes with a pair of double quotes above,
// or if the string includes a newline character (LF:\n:0x0A)
+ // or includes a carriage return (CR:\r:0x0D)
// and we are to allow newlines in fields,
// then the entire field must be enclosed in double quotes:
embeddedNewlineIndex = fieldStr.indexOf(LINEFEED);
+ embeddedCarriageReturn = fieldStr.indexOf(CARRIAGE_RETURN);
if ((fieldStr.indexOf(fieldDelimiter) != -1) ||
(fieldStr.indexOf(DOUBLE_QUOTE) != -1) ||
- (multilineTreatment == Multiline.YES) && (embeddedNewlineIndex != -1)) {
+ (multilineTreatment == Multiline.YES) && (embeddedNewlineIndex != -1 || embeddedCarriageReturn != -1)) {
fieldStr = "\"" + fieldStr + "\"";
}
Modified: pig/branches/branch-0.14/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java (original)
+++ pig/branches/branch-0.14/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java Wed Oct 1 06:43:56 2014
@@ -58,6 +58,7 @@ public class TestCSVExcelStorage {
"1st Field,\"A poem that continues\n" +
"for several lines\n" +
"do we\n" +
+ "(even with \r)" +
"handle that?\",Good,Fairy\n";
String[] testStrCommaArray =
@@ -69,7 +70,7 @@ public class TestCSVExcelStorage {
"\"Conrad\nEmil\",Dinger,40",
"Emil,\"\nDinger\",40",
"Quote problem,\"My \"\"famous\"\"\nsong\",60",
- "1st Field,\"A poem that continues\nfor several lines\ndo we\nhandle that?\",Good,Fairy",
+ "1st Field,\"A poem that continues\nfor several lines\ndo we\n(even with \r)handle that?\",Good,Fairy",
};
@SuppressWarnings("serial")
@@ -83,7 +84,7 @@ public class TestCSVExcelStorage {
add(Util.createTuple(new String[] {"Conrad\nEmil", "Dinger", "40"}));
add(Util.createTuple(new String[] {"Emil", "\nDinger", "40"}));
add(Util.createTuple(new String[] {"Quote problem", "My \"famous\"\nsong", "60"}));
- add(Util.createTuple(new String[] {"1st Field", "A poem that continues\nfor several lines\ndo we\nhandle that?", "Good", "Fairy"}));
+ add(Util.createTuple(new String[] {"1st Field", "A poem that continues\nfor several lines\ndo we\n(even with \r)handle that?", "Good", "Fairy"}));
}
};
@@ -104,7 +105,7 @@ public class TestCSVExcelStorage {
add(Util.createTuple(new String[] {"1st Field", "A poem that continues"}));
add(Util.createTuple(new String[] {"for several lines"}));
add(Util.createTuple(new String[] {"do we"}));
- add(Util.createTuple(new String[] {"handle that?,Good,Fairy"})); // Trailing double quote eats rest of line
+ add(Util.createTuple(new String[] {"(even with \r)handle that?,Good,Fairy"})); // Trailing double quote eats rest of line
}
};
@@ -161,6 +162,7 @@ public class TestCSVExcelStorage {
"1,,,,\"",
"qwe",
"rty\", uiop",
+ "1,10,2.718,3.14159,qwe\rrty,uiop",
"1,,,,\"qwe,rty\",uiop",
"1,,,,\"q\"\"wert\"\"y\", uiop",
"1,,,,qwerty,\"u\"\"io\"\"p\""
@@ -338,6 +340,7 @@ public class TestCSVExcelStorage {
"(1,10,,3.15159,,uiop)", // extra field (input data has "moose" after "uiop")
"(1,,2.718,,qwerty,uiop)", // quoted regular fields (2.718, qwerty, and uiop in quotes)
"(1,,,,\nqwe\nrty, uiop)", // newlines in quotes
+ "(1,10,2.718,3.14159,qwe\rrty,uiop)",// \r in text
"(1,,,,qwe,rty,uiop)", // commas in quotes
"(1,,,,q\"wert\"y, uiop)", // quotes in quotes
"(1,,,,qwerty,u\"io\"p)" // quotes in quotes at the end of a line
@@ -384,6 +387,7 @@ public class TestCSVExcelStorage {
"(1,,,,\")", // since we are just using TextLoader for verification
"(qwe)", // it treats the linebreaks as meaning separate records
"(rty\", uiop)", // but as shown in the load() test, CSVExcelStorage will read these properly
+ "1,10,2.718,3.14159,\"qwe\rrty\",uiop", // If \r is found text will be quoted
"(1,,,,\"qwe,rty\",uiop)",
"(1,,,,\"q\"\"wert\"\"y\", uiop)",
"(1,,,,qwerty,\"u\"\"io\"\"p\")"
Modified: pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Wed Oct 1 06:43:56 2014
@@ -640,16 +640,11 @@ public class JobControlCompiler{
for (String udf : mro.UDFs) {
if (udf.contains("GFCross")) {
- Object func = pigContext.instantiateFuncFromSpec(new FuncSpec(udf));
+ Object func = PigContext.instantiateFuncFromSpec(new FuncSpec(udf));
if (func instanceof GFCross) {
String crossKey = ((GFCross)func).getCrossKey();
- // If non GFCross has been processed yet
- if (pigContext.getProperties().get(PigConfiguration.PIG_CROSS_PARALLELISM_HINT + "." + crossKey)==null) {
- pigContext.getProperties().setProperty(PigConfiguration.PIG_CROSS_PARALLELISM_HINT + "." + crossKey,
- Integer.toString(nwJob.getNumReduceTasks()));
- }
conf.set(PigConfiguration.PIG_CROSS_PARALLELISM_HINT + "." + crossKey,
- (String)pigContext.getProperties().get(PigConfiguration.PIG_CROSS_PARALLELISM_HINT + "." + crossKey));
+ Integer.toString(mro.getRequestedParallelism()));
}
}
}
Modified: pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java (original)
+++ pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java Wed Oct 1 06:43:56 2014
@@ -1076,7 +1076,14 @@ public class MRCompiler extends PhyPlanV
@Override
public void visitPOForEach(POForEach op) throws VisitorException{
try{
- nonBlocking(op);
+ if (op.isMapSideOnly() && curMROp.isMapDone()) {
+ FileSpec fSpec = getTempFileSpec();
+ MapReduceOper prevMROper = endSingleInputPlanWithStr(fSpec);
+ curMROp = startNew(fSpec, prevMROper);
+ curMROp.mapPlan.addAsLeaf(op);
+ } else {
+ nonBlocking(op);
+ }
List<PhysicalPlan> plans = op.getInputPlans();
if(plans!=null)
for (PhysicalPlan plan : plans) {
Modified: pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java (original)
+++ pig/branches/branch-0.14/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java Wed Oct 1 06:43:56 2014
@@ -92,6 +92,10 @@ public class POForEach extends PhysicalO
protected Tuple inpTuple;
+ // Indicate the foreach statement can only in map side
+ // Currently only used in MR cross (See PIG-4175)
+ protected boolean mapSideOnly = false;
+
private Schema schema;
public POForEach(OperatorKey k) {
@@ -782,4 +786,12 @@ public class POForEach extends PhysicalO
return (Tuple) out;
}
}
+
+ public void setMapSideOnly(boolean mapSideOnly) {
+ this.mapSideOnly = mapSideOnly;
+ }
+
+ public boolean isMapSideOnly() {
+ return mapSideOnly;
+ }
}
Modified: pig/branches/branch-0.14/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java (original)
+++ pig/branches/branch-0.14/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java Wed Oct 1 06:43:56 2014
@@ -634,6 +634,7 @@ public class LogToPhyTranslationVisitor
List<PhysicalPlan> fePlans = Arrays.asList(fep1, fep2);
POForEach fe = new POForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), cross.getRequestedParallelism(), fePlans, flattenLst );
+ fe.setMapSideOnly(true);
fe.addOriginalLocation(cross.getAlias(), cross.getLocation());
currentPlan.add(fe);
currentPlan.connect(logToPhyMap.get(op), fe);
Modified: pig/branches/branch-0.14/test/org/apache/pig/test/TestEvalPipeline2.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.14/test/org/apache/pig/test/TestEvalPipeline2.java?rev=1628614&r1=1628613&r2=1628614&view=diff
==============================================================================
--- pig/branches/branch-0.14/test/org/apache/pig/test/TestEvalPipeline2.java (original)
+++ pig/branches/branch-0.14/test/org/apache/pig/test/TestEvalPipeline2.java Wed Oct 1 06:43:56 2014
@@ -31,10 +31,15 @@ import java.util.Map;
import java.util.Properties;
import java.util.Random;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
import org.apache.pig.EvalFunc;
import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecJob;
import org.apache.pig.builtin.BinStorage;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
@@ -44,11 +49,14 @@ import org.apache.pig.data.DefaultBagFac
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.PigImplConstants;
+import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.impl.util.ObjectSerializer;
import org.apache.pig.test.utils.Identity;
+import org.apache.pig.tools.pigstats.JobStats;
+import org.apache.pig.tools.pigstats.mapreduce.MRJobStats;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Assume;
@@ -1604,4 +1612,47 @@ public class TestEvalPipeline2 {
Assert.assertFalse(iter.hasNext());
}
+
+ @Test
+ public void testCrossAfterGroupAll() throws Exception{
+ String[] input = {
+ "1\tA",
+ "2\tB",
+ "3\tC",
+ "4\tD",
+ };
+
+ Util.createInputFile(cluster, "table_testCrossAfterGroupAll", input);
+
+ try {
+ pigServer.getPigContext().getProperties().setProperty("pig.exec.reducers.bytes.per.reducer", "40");
+ pigServer.registerQuery("A = load 'table_testCrossAfterGroupAll' as (a0, a1);");
+ pigServer.registerQuery("B = group A all;");
+ pigServer.registerQuery("C = foreach B generate COUNT(A);");
+ pigServer.registerQuery("D = cross A, C;");
+ Path output = FileLocalizer.getTemporaryPath(pigServer.getPigContext());
+ ExecJob job = pigServer.store("D", output.toString());
+ FileSystem fs = output.getFileSystem(cluster.getConfiguration());
+ FileStatus[] partFiles = fs.listStatus(output, new PathFilter() {
+ @Override
+ public boolean accept(Path path) {
+ if (path.getName().startsWith("part")) {
+ return true;
+ }
+ return false;
+ }
+ });
+ // auto-parallelism is 2 in MR, 20 in Tez, so check >=2
+ Assert.assertTrue(partFiles.length >= 2);
+ // Check the count of output
+ Iterator<Tuple> iter = job.getResults();
+ iter.next();
+ iter.next();
+ iter.next();
+ iter.next();
+ Assert.assertFalse(iter.hasNext());
+ } finally {
+ pigServer.getPigContext().getProperties().remove("pig.exec.reducers.bytes.per.reducer");
+ }
+ }
}