diff --git a/CHANGES.txt b/CHANGES.txt
index 6a267c2..f56a414 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -32,6 +32,8 @@
  
 BUG FIXES
 
+PIG-4548: Records Lost With Specific Combination of Commands and Streaming Function (knoguchi)
+
 PIG-5262: Fix jdiff related issues: fail build upon error, correct xml character escaping (szita)
 
 PIG-5225: Several unit tests are not annotated with @Test (nkollar via rohini)
diff --git a/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODemux.java b/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODemux.java
index f7efbe7..1e34264 100644
--- a/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODemux.java
+++ b/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/PODemux.java
@@ -272,7 +272,19 @@
                 }
             } else {
                 curLeaf = leaf;
-                res = leaf.getNextTuple();
+                while (true) {
+                    res = leaf.getNextTuple();
+
+                    if (res.returnStatus == POStatus.STATUS_OK ||
+                            res.returnStatus == POStatus.STATUS_EOP ||
+                            res.returnStatus == POStatus.STATUS_ERR) {
+                        break;
+                    } else if (res.returnStatus == POStatus.STATUS_NULL) {
+                        // this "else if" is unnecessary but keeping it to
+                        // be sync with runPipeline()
+                        continue;
+                    }
+                }
                
                 if (res.returnStatus == POStatus.STATUS_EOP)  {
                     processedSet.set(idx++);        
diff --git a/test/org/apache/pig/test/TestMultiQuery.java b/test/org/apache/pig/test/TestMultiQuery.java
index 47cc8ed..3612fe4 100644
--- a/test/org/apache/pig/test/TestMultiQuery.java
+++ b/test/org/apache/pig/test/TestMultiQuery.java
@@ -65,6 +65,14 @@
         deleteOutputFiles();
     }
 
+    private static final String simpleEchoStreamingCommand;
+    static {
+        if (Util.WINDOWS)
+            simpleEchoStreamingCommand = "perl -ne 'print \\\"$_\\\"'";
+        else
+            simpleEchoStreamingCommand = "perl -ne 'print \"$_\"'";
+    }
+
     @Before
     public void setUp() throws Exception {
         deleteOutputFiles();
@@ -955,6 +963,40 @@
                 .translateSchema(myPig.dumpSchema("C2")), Util.isSparkExecType(Util.getLocalTestMode()));
     }
 
+    @Test
+    public void testMultiQueryJiraPig4548() throws Exception {
+        Storage.Data data = Storage.resetData(myPig);
+        data.set("inputLocation",
+                Storage.tuple("1", "f"),
+                Storage.tuple("2", "f"),
+                Storage.tuple("3", "f"),
+                Storage.tuple("4", "f"),
+                Storage.tuple("5", "f"),
+                Storage.tuple("6", "f"));
+        myPig.setBatchOn();
+        myPig.registerQuery("A = load 'inputLocation' using mock.Storage() as (f1:chararray, f2:chararray);");
+        myPig.registerQuery("SPLIT A into T if (f2 == 'T'), F otherwise;");
+        myPig.registerQuery("T2 = group T by f1;");
+        myPig.registerQuery("store T2 into 'output1' using mock.Storage();");
+        myPig.registerQuery("F2 = group F by f1;");
+        myPig.registerQuery("F3 = stream F2 through `" + simpleEchoStreamingCommand
+                            + "` as (group:chararray, F:bag{tuple(f1: chararray,f2: chararray)});");
+        myPig.registerQuery("store F3 into 'output2' using mock.Storage();");
+        myPig.executeBatch();
+
+        List<Tuple> actualResults = data.get("output1");
+        assertEquals("Number of records for output1 should be 0",0,actualResults.size());
+        actualResults = data.get("output2");
+        List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
+                                         new String[]{"('1',{('1','f')})",
+                                                      "('2',{('2','f')})",
+                                                      "('3',{('3','f')})",
+                                                      "('4',{('4','f')})",
+                                                      "('5',{('5','f')})",
+                                                      "('6',{('6','f')})"});
+        Util.checkQueryOutputsAfterSort(actualResults.iterator(), expectedResults);
+    }
+
     // --------------------------------------------------------------------------
     // Helper methods