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 2011/03/23 01:49:13 UTC
svn commit: r1084429 - in /pig/branches/branch-0.8: ./
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/newplan/logical/relational/
src/org/apache/pig/newplan/logical/rules/ test/org/apache/pig/test/
Author: daijy
Date: Wed Mar 23 00:49:12 2011
New Revision: 1084429
URL: http://svn.apache.org/viewvc?rev=1084429&view=rev
Log:
PIG-1912: non-deterministic output when a file is loaded multiple times
Added:
pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/rules/LoadStoreFuncDupSignatureValidator.java
Modified:
pig/branches/branch-0.8/CHANGES.txt
pig/branches/branch-0.8/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LOLoad.java
pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
pig/branches/branch-0.8/test/org/apache/pig/test/TestEvalPipeline2.java
Modified: pig/branches/branch-0.8/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.8/CHANGES.txt?rev=1084429&r1=1084428&r2=1084429&view=diff
==============================================================================
--- pig/branches/branch-0.8/CHANGES.txt (original)
+++ pig/branches/branch-0.8/CHANGES.txt Wed Mar 23 00:49:12 2011
@@ -227,6 +227,8 @@ PIG-1309: Map-side Cogroup (ashutoshc)
BUG FIXES
+PIG-1912: non-deterministic output when a file is loaded multiple times (daijy)
+
PIG-1892: Bug in new logical plan : No output generated even though there are
valid records (daijy)
Modified: pig/branches/branch-0.8/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.8/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java?rev=1084429&r1=1084428&r2=1084429&view=diff
==============================================================================
--- pig/branches/branch-0.8/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java (original)
+++ pig/branches/branch-0.8/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Wed Mar 23 00:49:12 2011
@@ -77,6 +77,7 @@ import org.apache.pig.newplan.logical.re
import org.apache.pig.newplan.logical.relational.LOStore;
import org.apache.pig.newplan.logical.relational.LogicalRelationalNodesVisitor;
import org.apache.pig.newplan.logical.rules.InputOutputFileValidator;
+import org.apache.pig.newplan.logical.rules.LoadStoreFuncDupSignatureValidator;
import org.apache.pig.newplan.optimizer.Rule;
import org.apache.pig.tools.pigstats.OutputStats;
import org.apache.pig.tools.pigstats.PigStats;
@@ -271,6 +272,9 @@ public class HExecutionEngine {
throw new FrontendException(msg, errCode, PigException.BUG, ioe);
}
+ LoadStoreFuncDupSignatureValidator loadStoreFuncDupSignatureValidator = new LoadStoreFuncDupSignatureValidator(newPlan);
+ loadStoreFuncDupSignatureValidator.validate();
+
// run optimizer
org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer optimizer =
new org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer(newPlan, 100, optimizerRules);
Modified: pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LOLoad.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LOLoad.java?rev=1084429&r1=1084428&r2=1084429&view=diff
==============================================================================
--- pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LOLoad.java (original)
+++ pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LOLoad.java Wed Mar 23 00:49:12 2011
@@ -45,6 +45,7 @@ public class LOLoad extends LogicalRelat
private List<Integer> requiredFields = null;
private boolean castInserted = false;
private LogicalSchema uidOnlySchema;
+ private String signature = null;
/**
*
@@ -64,7 +65,7 @@ public class LOLoad extends LogicalRelat
try {
if (loadFunc == null && fs!=null) {
loadFunc = (LoadFunc)PigContext.instantiateFuncFromSpec(fs.getFuncSpec());
- loadFunc.setUDFContextSignature(getAlias());
+ loadFunc.setUDFContextSignature(signature);
}
return loadFunc;
@@ -205,4 +206,24 @@ public class LOLoad extends LogicalRelat
String str = super.toString();
return (str + "RequiredFields:" + requiredFields);
}
+
+ public String getSignature() {
+ return signature;
+ }
+
+ public void setAlias(String alias) {
+ super.setAlias(alias);
+ if (signature==null)
+ signature = alias;
+ }
+
+ /***
+ * This method is called by Pig logical planner to setup UDFContext signature.
+ * So that loadFunc can use signature to store its own configurations in UDFContext.
+ * This is not intend to be called by users
+ */
+ public void setSignature(String signature) {
+ this.signature = signature;
+ loadFunc.setUDFContextSignature(signature);
+ }
}
Modified: pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java?rev=1084429&r1=1084428&r2=1084429&view=diff
==============================================================================
--- pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java (original)
+++ pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java Wed Mar 23 00:49:12 2011
@@ -121,7 +121,7 @@ public class LogToPhyTranslationVisitor
load.setLFile(loLoad.getFileSpec());
load.setPc(pc);
load.setResultType(DataType.BAG);
- load.setSignature(loLoad.getAlias());
+ load.setSignature(loLoad.getSignature());
currentPlan.add(load);
logToPhyMap.put(loLoad, load);
Added: pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/rules/LoadStoreFuncDupSignatureValidator.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/rules/LoadStoreFuncDupSignatureValidator.java?rev=1084429&view=auto
==============================================================================
--- pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/rules/LoadStoreFuncDupSignatureValidator.java (added)
+++ pig/branches/branch-0.8/src/org/apache/pig/newplan/logical/rules/LoadStoreFuncDupSignatureValidator.java Wed Mar 23 00:49:12 2011
@@ -0,0 +1,105 @@
+/*
+ * 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.newplan.logical.rules;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.newplan.DepthFirstWalker;
+import org.apache.pig.newplan.OperatorPlan;
+import org.apache.pig.newplan.logical.relational.LOLoad;
+import org.apache.pig.newplan.logical.relational.LOStore;
+import org.apache.pig.newplan.logical.relational.LogicalRelationalNodesVisitor;
+
+/***
+ * Check for duplicate alias. This is because we use alias as LoadFunc signature. If
+ * user use the same alias in two load statement, we cannot distinguish two loadFunc.
+ * LoadStoreFuncDupSignatureValidator solve this problem by appending an index if signature
+ * conflicts
+ */
+
+public class LoadStoreFuncDupSignatureValidator {
+ OperatorPlan plan;
+ public LoadStoreFuncDupSignatureValidator(OperatorPlan plan) {
+ this.plan = plan;
+ }
+
+ public void validate() throws FrontendException {
+ LoadStoreFuncDupSignatureVisitor visitor = new LoadStoreFuncDupSignatureVisitor(plan);
+ visitor.visit();
+ visitor.finish();
+ }
+
+ static class LoadStoreFuncDupSignatureVisitor extends LogicalRelationalNodesVisitor {
+ Map<String, List<LOLoad>> loadSignatures = new HashMap<String, List<LOLoad>>();
+ Map<String, List<LOStore>> storeSignatures = new HashMap<String, List<LOStore>>();
+ protected LoadStoreFuncDupSignatureVisitor(OperatorPlan plan)
+ throws FrontendException {
+ super(plan, new DepthFirstWalker(plan));
+ }
+
+ @Override
+ public void visit(LOLoad load) throws FrontendException {
+ if (loadSignatures.containsKey(load.getSignature())) {
+ List<LOLoad> loads = loadSignatures.get(load.getSignature());
+ loads.add(load);
+ } else {
+ List<LOLoad> loads = new ArrayList<LOLoad>();
+ loads.add(load);
+ loadSignatures.put(load.getSignature(), loads);
+ }
+ }
+
+ @Override
+ public void visit(LOStore store) throws FrontendException {
+ if (storeSignatures.containsKey(store.getSignature())) {
+ List<LOStore> stores = storeSignatures.get(store.getSignature());
+ stores.add(store);
+ } else {
+ List<LOStore> stores = new ArrayList<LOStore>();
+ stores.add(store);
+ storeSignatures.put(store.getSignature(), stores);
+ }
+ }
+
+ public void finish() {
+ for (Map.Entry<String, List<LOLoad>> entry : loadSignatures.entrySet()) {
+ String key = entry.getKey();
+ List<LOLoad> loads = entry.getValue();
+ if (loads.size()>1) {
+ for (int i=0;i<loads.size();i++) {
+ loads.get(i).setSignature(key+"$"+i);
+ }
+ }
+ }
+
+ for (Map.Entry<String, List<LOStore>> entry : storeSignatures.entrySet()) {
+ String key = entry.getKey();
+ List<LOStore> stores = entry.getValue();
+ if (stores.size()>1) {
+ for (int i=0;i<stores.size();i++) {
+ stores.get(i).setSignature(key+"$"+i);
+ }
+ }
+ }
+ }
+ }
+}
Modified: pig/branches/branch-0.8/test/org/apache/pig/test/TestEvalPipeline2.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.8/test/org/apache/pig/test/TestEvalPipeline2.java?rev=1084429&r1=1084428&r2=1084429&view=diff
==============================================================================
--- pig/branches/branch-0.8/test/org/apache/pig/test/TestEvalPipeline2.java (original)
+++ pig/branches/branch-0.8/test/org/apache/pig/test/TestEvalPipeline2.java Wed Mar 23 00:49:12 2011
@@ -1227,4 +1227,38 @@ public class TestEvalPipeline2 extends T
assertTrue(t.toString().equals("({(1),(2)})"));
assertFalse(iter.hasNext());
}
+
+ // See PIG-1912
+ @Test
+ public void testDuplicateLoadFuncSignature() throws Exception{
+ String[] input = {
+ "0\t1\ta",
+ };
+
+ Util.createInputFile(cluster, "table_testDuplicateLoadFuncSignature", input);
+ pigServer.setBatchOn();
+ pigServer.registerQuery("a = load 'table_testDuplicateLoadFuncSignature' as (a0, a1, a2);");
+ pigServer.registerQuery("b = foreach a generate a0, a1;");
+ pigServer.registerQuery("a = load 'table_testDuplicateLoadFuncSignature' as (a0, a1, a2);");
+ pigServer.registerQuery("c = foreach a generate a0, a2;");
+ pigServer.registerQuery("store b into 'testDuplicateLoadFuncSignatureOutput1';");
+ pigServer.registerQuery("store c into 'testDuplicateLoadFuncSignatureOutput2';");
+
+ pigServer.executeBatch();
+
+ pigServer.registerQuery("a = load 'testDuplicateLoadFuncSignatureOutput1';");
+ Iterator<Tuple> iter = pigServer.openIterator("a");
+
+ Tuple t = iter.next();
+ assertTrue(t.toString().equals("(0,1)"));
+ assertFalse(iter.hasNext());
+
+ pigServer.registerQuery("a = load 'testDuplicateLoadFuncSignatureOutput2';");
+ iter = pigServer.openIterator("a");
+
+ t = iter.next();
+ assertTrue(t.toString().equals("(0,a)"));
+ assertFalse(iter.hasNext());
+
+ }
}