diff --git a/CHANGES.txt b/CHANGES.txt
index d240afa..fc51596 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -88,6 +88,8 @@
  
 BUG FIXES
 
+PIG-5370: Union onschema + columnprune dropping used fields (knoguchi)
+
 PIG-5362: Parameter substitution of shell cmd results doesn't handle backslash (wlauer via rohini)
 
 PIG-5355: Negative progress report by HBaseTableRecordReader (satishsaley via knoguchi)
diff --git a/src/org/apache/pig/newplan/logical/relational/LOUnion.java b/src/org/apache/pig/newplan/logical/relational/LOUnion.java
index fbd42ed..137f038 100644
--- a/src/org/apache/pig/newplan/logical/relational/LOUnion.java
+++ b/src/org/apache/pig/newplan/logical/relational/LOUnion.java
@@ -34,9 +34,11 @@
 
 public class LOUnion extends LogicalRelationalOperator {
     private boolean onSchema;
+
+    private static String UID_SEPARATOR = "_";
     
     // uid mapping from output uid to input uid
-    private List<Pair<Long, Long>> uidMapping = new ArrayList<Pair<Long, Long>>();
+    private List<Pair<Long, String>> uidMapping = new ArrayList<Pair<Long, String>>();
     
     public LOUnion(OperatorPlan plan) {
         super("LOUnion", plan);
@@ -108,7 +110,7 @@
         }
 
         // Bring back cached uid if any; otherwise, cache uid generated
-        setMergedSchemaUids(mergedSchema, inputSchemas);
+        setMergedSchemaUids(mergedSchema, inputSchemas, "");
 
         return schema = mergedSchema;
     }
@@ -145,7 +147,7 @@
         return mergedSchema;
     }
 
-    private void setMergedSchemaUids(LogicalSchema mergedSchema, List<LogicalSchema> inputSchemas)
+    private void setMergedSchemaUids(LogicalSchema mergedSchema, List<LogicalSchema> inputSchemas, String nested_uids)
     throws FrontendException {
 
         for (int i=0;i<mergedSchema.size();i++) {
@@ -170,7 +172,7 @@
                     }
 
                     if (uid < 0) {
-                        uid = getCachedOuputUid(inputFieldSchema.uid);
+                        uid = getCachedOuputUid(createNestedUids(nested_uids,inputFieldSchema.uid));
                         if (uid >= 0 && outputFieldSchema.schema == null) break;
                     }
                 }
@@ -186,13 +188,13 @@
                         matchedInputFieldSchema = inputSchema.getFieldSubNameMatch(mergedSchema.getField(i).alias);
                         if (matchedInputFieldSchema!=null) {
                             inputUid = matchedInputFieldSchema.uid;
-                            uidMapping.add(new Pair<Long, Long>(uid, inputUid));
+                            uidMapping.add(new Pair<Long, String>(uid, createNestedUids(nested_uids,inputUid)));
                         }
                     }
                     else {
                         matchedInputFieldSchema = mergedSchema.getField(i);
                         inputUid = inputSchema.getField(i).uid;
-                        uidMapping.add(new Pair<Long, Long>(uid, inputUid));
+                        uidMapping.add(new Pair<Long, String>(uid, createNestedUids(nested_uids,inputUid)));
                     }
                 }
             }
@@ -201,16 +203,28 @@
 
             // This field has a schema. Assign uids to it as well
             if (outputFieldSchema.schema != null) {
-                setMergedSchemaUids(outputFieldSchema.schema, fieldInputSchemas);
+                setMergedSchemaUids(outputFieldSchema.schema, fieldInputSchemas, createNestedUids(nested_uids,outputFieldSchema.uid));
             }
         }
     }
 
-    private long getCachedOuputUid(long inputUid) {
+    private String createNestedUids(String nested_uids, long new_uid) {
+        StringBuilder sb = new StringBuilder(nested_uids);
+        sb.append(UID_SEPARATOR);
+        sb.append(new_uid);
+        return sb.toString();
+    }
+
+    private long getLeafUid(String nested_uids) {
+        String [] uid_root_to_leaf = nested_uids.split(UID_SEPARATOR);
+        return Long.valueOf(uid_root_to_leaf[uid_root_to_leaf.length-1]);
+    }
+
+    private long getCachedOuputUid(String nested_input_uids) {
         long uid = -1;
         
-        for (Pair<Long, Long> pair : uidMapping) {
-            if (pair.second==inputUid) {
+        for (Pair<Long, String> pair : uidMapping) {
+            if (pair.second.equals(nested_input_uids)) {
                 uid = pair.first;
                 break;
             }
@@ -237,18 +251,18 @@
     }
 
     // Get input uids mapping to the output uid
-    public Set<Long> getInputUids(long uid) {
+    public Set<Long> getInputUids(long outputuid) {
         Set<Long> result = new HashSet<Long>();
-        for (Pair<Long, Long> pair : uidMapping) {
-            if (pair.first==uid)
-                result.add(pair.second);
+        for (Pair<Long, String> pair : uidMapping) {
+            if (pair.first==outputuid)
+                result.add(getLeafUid(pair.second));
         }
         return result;
     }
     
     @Override
     public void resetUid() {
-        uidMapping = new ArrayList<Pair<Long, Long>>();
+        uidMapping = new ArrayList<Pair<Long, String>>();
     }
     
     public List<Operator> getInputs() {
diff --git a/test/org/apache/pig/test/TestNewPlanColumnPrune.java b/test/org/apache/pig/test/TestNewPlanColumnPrune.java
index 67a02af..8193bf6 100644
--- a/test/org/apache/pig/test/TestNewPlanColumnPrune.java
+++ b/test/org/apache/pig/test/TestNewPlanColumnPrune.java
@@ -482,6 +482,35 @@
         }
     }
 
+    @Test
+    public void testUnionOnschemaWithInnerBag() throws Exception  {
+        // After handing inner-bag in Union-onschema,
+        // ColumnPrune broke due to overlapping uid inside the relation and
+        // ones inside the inner-bag  (PIG-5370)
+        String query = "A0 = load 'd.txt' as (a0:int, a1:int, a2:int, a3:int);" +
+        "A = FOREACH A0 GENERATE a0, a1, a2;" +
+        "B = FOREACH (GROUP A by (a0,a1)) {" +
+        "    A_FOREACH = FOREACH A GENERATE a1,a2;" +
+        "    GENERATE A, FLATTEN(A_FOREACH) as (a1,a2);" +
+        "}" +
+        "C = load 'd2.txt' as (A:bag{tuple:(a0:int, a1:int, a2:int)}, a1:int,a2:int);" +
+        "Z = UNION ONSCHEMA B, C;"  +
+        "store Z into 'empty';";
+
+        LogicalPlan newLogicalPlan = buildPlan(query);
+
+        PlanOptimizer optimizer = new MyPlanOptimizer(newLogicalPlan, 3);
+        optimizer.optimize();
+        System.err.println(newLogicalPlan);
+        Iterator<Operator> iter = newLogicalPlan.getOperators();
+        while (iter.hasNext()) {
+            Operator o = iter.next();
+            LogicalRelationalOperator lro = (LogicalRelationalOperator)o;
+            if (lro == null || lro.getAlias() == null) continue;
+            assertNotNull(lro.getSchema());
+        }
+    }
+
     public class MyPlanOptimizer extends LogicalPlanOptimizer {
 
         protected MyPlanOptimizer(OperatorPlan p,  int iterations) {
@@ -505,4 +534,3 @@
         }
     }
 }
-