svn commit: r1295452 - in /pig/trunk: CHANGES.txt src/org/apache/pig/newplan/logical/rules/MapKeysPruneHelper.java test/org/apache/pig/test/TestPruneColumn.java

[da...@apache.org]

Author: daijy
Date: Thu Mar  1 07:56:48 2012
New Revision: 1295452

URL: http://svn.apache.org/viewvc?rev=1295452&view=rev
Log:
PIG-2535: Bug in new logical plan results in no output for join

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/newplan/logical/rules/MapKeysPruneHelper.java
    pig/trunk/test/org/apache/pig/test/TestPruneColumn.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1295452&r1=1295451&r2=1295452&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Thu Mar  1 07:56:48 2012
@@ -417,6 +417,8 @@ Release 0.9.3 - Unreleased
 
 BUG FIXES
 
+PIG-2535: Bug in new logical plan results in no output for join (daijy)
+
 PIG-2534: Pig generating infinite map outputs (daijy)
 
 PIG-2493: UNION causes casting issues (vivekp via daijy)

Modified: pig/trunk/src/org/apache/pig/newplan/logical/rules/MapKeysPruneHelper.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/rules/MapKeysPruneHelper.java?rev=1295452&r1=1295451&r2=1295452&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/rules/MapKeysPruneHelper.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/rules/MapKeysPruneHelper.java Thu Mar  1 07:56:48 2012
@@ -260,7 +260,10 @@ public class MapKeysPruneHelper {
         
         @Override
         public void visit(LOSplitOutput splitOutput) throws FrontendException {
-            super.visit(splitOutput);
+            currentOp = splitOutput;
+            MapExprMarker v = (MapExprMarker) getVisitor(splitOutput.getFilterPlan());
+            v.visit();
+            mergeUidKeys( v.inputUids );
             if (splitOutput.getSchema()!=null) {
                 for (LogicalFieldSchema fs : splitOutput.getSchema().getFields()) {
                     long inputUid = splitOutput.getInputUids(fs.uid);

Modified: pig/trunk/test/org/apache/pig/test/TestPruneColumn.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestPruneColumn.java?rev=1295452&r1=1295451&r2=1295452&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestPruneColumn.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestPruneColumn.java Thu Mar  1 07:56:48 2012
@@ -2102,4 +2102,29 @@ public class TestPruneColumn extends Tes
         assertTrue(checkLogFileMessage(new String[]{"Map key required for event_serve: $0->[key4, key3]", 
                 "Map key required for cm_data_raw: $0->[key4, key3, key5]"}));
     }
+
+    // See PIG-2535
+    @Test
+    public void testStream3() throws Exception {
+
+        pigServer.registerQuery("event_serve = LOAD 'input1' AS (s, m, l);");
+        pigServer.registerQuery("raw = LOAD 'input2' AS (s, m, l);");
+
+        pigServer.registerQuery("SPLIT raw INTO " +
+            "serve_raw IF (( (chararray) (s#'type') == '0') AND ( (chararray) (s#'source') == '5'))," +
+            "cm_click_raw IF (( (chararray) (s#'type') == '1') AND ( (chararray) (s#'source') == '5'));");
+        pigServer.registerQuery("cm_serve = FOREACH serve_raw GENERATE  s#'cm_serve_id' AS cm_event_guid,  s#'cm_serve_timestamp_ms' AS cm_receive_time, s#'p_url' AS ctx ;");
+        pigServer.registerQuery("cm_serve_lowercase = stream cm_serve through `tr [:upper:] [:lower:]`;");
+        pigServer.registerQuery("cm_serve_final = FOREACH cm_serve_lowercase GENERATE  $0 AS cm_event_guid, $1 AS cm_receive_time, $2 AS ctx;");
+        pigServer.registerQuery("filtered = FILTER event_serve BY (chararray) (s#'filter_key') neq 'xxxx' AND (chararray) (s#'filter_key') neq 'yyyy';");
+        pigServer.registerQuery("event_serve_project = FOREACH filtered GENERATE s#'event_guid' AS event_guid, s#'receive_time' AS receive_time;");
+        pigServer.registerQuery("event_serve_join = join cm_serve_final by (cm_event_guid), event_serve_project by (event_guid);");
+
+
+        pigServer.explain("event_serve_join", System.out);
+
+        assertTrue(checkLogFileMessage(new String[]{"Map key required for event_serve: $0->[event_guid, receive_time, filter_key]",
+                "Map key required for raw: $0->[source, p_url, cm_serve_timestamp_ms, cm_serve_id, type]"}));
+    }
+
 }