pig-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dvrya...@apache.org
Subject svn commit: r1391584 - in /pig/trunk: ./ src/org/apache/pig/data/ src/org/apache/pig/impl/util/ test/org/apache/pig/test/
Date Fri, 28 Sep 2012 18:30:23 GMT
Author: dvryaboy
Date: Fri Sep 28 18:30:22 2012
New Revision: 1391584

URL: http://svn.apache.org/viewvc?rev=1391584&view=rev
Log:
PIG-2923: Lazily register bags with SpillableMemoryManager

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/data/BagFactory.java
    pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java
    pig/trunk/src/org/apache/pig/data/DefaultBagFactory.java
    pig/trunk/src/org/apache/pig/data/DistinctDataBag.java
    pig/trunk/src/org/apache/pig/data/InternalCachedBag.java
    pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java
    pig/trunk/src/org/apache/pig/data/InternalSortedBag.java
    pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java
    pig/trunk/test/org/apache/pig/test/TestDataBag.java
    pig/trunk/test/org/apache/pig/test/TestDataBagAccess.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Fri Sep 28 18:30:22 2012
@@ -25,8 +25,10 @@ PIG-1891 Enable StoreFunc to make intell
 
 IMPROVEMENTS
 
-PIG-2929: Improve documentation around AVG, CONCAT, MIN, MAX (cheolsoo via billgraham)
+PIG-2923: Lazily register bags with SpillableMemoryManager (dvryaboy)
 
+PIG-2929: Improve documentation around AVG, CONCAT, MIN, MAX (cheolsoo via billgraham)
+ 
 PIG-2852: Update documentation regarding parallel local mode execution (cheolsoo via jcoveney)
 
 PIG-2879: Pig current releases lack a UDF startsWith.This UDF tests if a given string starts
with the specified prefix. (initialcontext via azaroth)

Modified: pig/trunk/src/org/apache/pig/data/BagFactory.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/BagFactory.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/BagFactory.java (original)
+++ pig/trunk/src/org/apache/pig/data/BagFactory.java Fri Sep 28 18:30:22 2012
@@ -120,10 +120,11 @@ public abstract class BagFactory {
      * Construct a new BagFactory
      */
     protected BagFactory() {
-        gMemMgr = new SpillableMemoryManager();
+        gMemMgr = SpillableMemoryManager.getInstance();
     }
 
     /**
+     * @deprecated As of Pig 0.11, bags register with the {@link SpillableMemoryManager}
themselves.
      * Register a bag with the
      * {@link org.apache.pig.impl.util.SpillableMemoryManager}.
      * If the bags created by an implementation of BagFactory are managed by
@@ -131,6 +132,7 @@ public abstract class BagFactory {
      * method should be called each time a new bag is created.
      * @param b bag to be registered.
      */
+    @Deprecated
     protected void registerBag(DataBag b) {
         gMemMgr.registerSpillable(b);
     }

Modified: pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/DefaultAbstractBag.java Fri Sep 28 18:30:22 2012
@@ -18,17 +18,17 @@
 package org.apache.pig.data;
 
 import java.io.BufferedOutputStream;
+import java.io.DataInput;
+import java.io.DataOutput;
 import java.io.DataOutputStream;
 import java.io.File;
 import java.io.FileOutputStream;
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Collection;
 import java.util.Iterator;
-import java.util.ArrayList;
 
-import org.apache.pig.PigCounters;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.pig.PigException;
 import org.apache.pig.PigWarning;
 import org.apache.pig.backend.executionengine.ExecException;
@@ -36,19 +36,21 @@ import org.apache.pig.backend.hadoop.exe
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PigLogger;
 import org.apache.pig.impl.util.BagFormat;
-import org.apache.pig.impl.util.Spillable;
+import org.apache.pig.impl.util.SpillableMemoryManager;
 import org.apache.pig.tools.pigstats.PigStatusReporter;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 
 /**
  * Default implementation of DataBag.  This is the an abstract class used as a
  * parent for all three of the types of data bags.
  */
+@SuppressWarnings("serial")
 public abstract class DefaultAbstractBag implements DataBag {
 
     private static final Log log = LogFactory.getLog(DataBag.class);
 
+    // If we grow past 100K, may be worthwhile to register.
+    private static final int SPILL_REGISTER_THRESHOLD = 100 * 1024;
+
     private static PigLogger pigLogger = PhysicalOperator.getPigLogger();
 
     private static InterSedes sedes = InterSedesFactory.getInterSedesInstance();
@@ -67,9 +69,12 @@ public abstract class DefaultAbstractBag
 
     protected long mMemSize = 0;
 
+    private boolean spillableRegistered = false;
+
     /**
      * Get the number of elements in the bag, both in memory and on disk.
      */
+    @Override
     public long size() {
         return mSize;
     }
@@ -78,40 +83,52 @@ public abstract class DefaultAbstractBag
      * Add a tuple to the bag.
      * @param t tuple to add.
      */
+    @Override
     public void add(Tuple t) {
         synchronized (mContents) {
             mSize++;
             mContents.add(t);
         }
+        markSpillableIfNecessary();
     }
 
     /**
-     * Add contents of a bag to the bag.
-     * @param b bag to add contents of.
+     * All bag implementations that can get big enough to be spilled
+     * should call this method after every time they add an element.
      */
+    protected void markSpillableIfNecessary() {
+        if (!spillableRegistered && getMemorySize() >= SPILL_REGISTER_THRESHOLD)
{
+            SpillableMemoryManager.getInstance().registerSpillable(this);
+            spillableRegistered = true;
+        }
+    }
+
+    @Override
     public void addAll(DataBag b) {
-        synchronized (mContents) {
-            mSize += b.size();
-            Iterator<Tuple> i = b.iterator();
-            while (i.hasNext()) mContents.add(i.next());
+        addAll((Iterable<Tuple>) b);
         }
+
+    public void addAll(Collection<Tuple> c) {
+        addAll((Iterable<Tuple>) c);
     }
 
     /**
-     * Add contents of a container to the bag.
-     * @param c Collection to add contents of.
+     * Add contents of an iterable (a collection or a DataBag)
+     *
+     * @param iterable a Collection or DataBag to add contents of
      */
-    public void addAll(Collection<Tuple> c) {
+    public void addAll(Iterable<Tuple> iterable) {
         synchronized (mContents) {
-            mSize += c.size();
-            Iterator<Tuple> i = c.iterator();
-            while (i.hasNext()) mContents.add(i.next());
+            for (Tuple t : iterable) {
+                add(t);
+            }
         }
     }
 
     /**
      * Return the size of memory usage.
      */
+    @Override
     public long getMemorySize() {
         int j;
         int numInMem = 0;
@@ -188,6 +205,7 @@ public abstract class DefaultAbstractBag
      * Any attempts to read after this is called will produce undefined
      * results.
      */
+    @Override
     public void clear() {
         synchronized (mContents) {
             mContents.clear();
@@ -207,6 +225,7 @@ public abstract class DefaultAbstractBag
      * This method is potentially very expensive since it may require a
      * sort of the bag; don't call it unless you have to.
      */
+    @Override
     @SuppressWarnings("unchecked")
     public int compareTo(Object other) {
         if (this == other)
@@ -268,6 +287,7 @@ public abstract class DefaultAbstractBag
      * @param out DataOutput to write data to.
      * @throws IOException (passes it on from underlying calls).
      */
+    @Override
     public void write(DataOutput out) throws IOException {
         sedes.writeDatum(out, this);
     }
@@ -277,6 +297,7 @@ public abstract class DefaultAbstractBag
      * @param in DataInput to read data from.
      * @throws IOException (passes it on from underlying calls).
      */
+    @Override
     public void readFields(DataInput in) throws IOException {
         long size = in.readLong();
         
@@ -294,6 +315,7 @@ public abstract class DefaultAbstractBag
      * This is used by FuncEvalSpec.FakeDataBag.
      * @param stale Set stale state.
      */
+    @Override
     public void markStale(boolean stale)
     {
     }
@@ -370,6 +392,7 @@ public abstract class DefaultAbstractBag
         }
     }
 
+    @SuppressWarnings("rawtypes")
     protected void warn(String msg, Enum warningEnum, Exception e) {
     	pigLogger = PhysicalOperator.getPigLogger();
     	if(pigLogger != null) {
@@ -379,10 +402,12 @@ public abstract class DefaultAbstractBag
     	}    	
     }
     
+    @SuppressWarnings("rawtypes")
     protected void incSpillCount(Enum counter) {
         incSpillCount(counter, 1);
     }
 
+    @SuppressWarnings("rawtypes")
     protected void incSpillCount(Enum counter, long numRecsSpilled) {
         PigStatusReporter reporter = PigStatusReporter.getInstance();
         if (reporter != null && reporter.getCounter(counter)!=null) {

Modified: pig/trunk/src/org/apache/pig/data/DefaultBagFactory.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/DefaultBagFactory.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/DefaultBagFactory.java (original)
+++ pig/trunk/src/org/apache/pig/data/DefaultBagFactory.java Fri Sep 28 18:30:22 2012
@@ -20,8 +20,6 @@ package org.apache.pig.data;
 import java.util.Comparator;
 import java.util.List;
 
-import org.apache.pig.impl.util.SpillableMemoryManager;
-
 /**
  * Default implementation of BagFactory.
  */
@@ -29,9 +27,9 @@ public class DefaultBagFactory extends B
     /**
      * Get a default (unordered, not distinct) data bag.
      */
+    @Override
     public DataBag newDefaultBag() {
         DataBag b = new DefaultDataBag();
-        registerBag(b);
         return b;
     }
     
@@ -41,9 +39,9 @@ public class DefaultBagFactory extends B
      * copy the tuples but uses the provided list as its backing store.
      * So it takes ownership of the list.
      */
+    @Override
     public DataBag newDefaultBag(List<Tuple> listOfTuples) {
         DataBag b = new DefaultDataBag(listOfTuples);
-        registerBag(b);
         return b;
     }
 
@@ -52,18 +50,18 @@ public class DefaultBagFactory extends B
      * @param comp Comparator that controls how the data is sorted.
      * If null, default comparator will be used.
      */
+    @Override
     public DataBag newSortedBag(Comparator<Tuple> comp) {
         DataBag b = new SortedDataBag(comp);
-        registerBag(b);
         return b;
     }
     
     /**
      * Get a distinct data bag.
      */
+    @Override
     public DataBag newDistinctBag() {
         DataBag b = new DistinctDataBag();
-        registerBag(b);
         return b;
     }
 

Modified: pig/trunk/src/org/apache/pig/data/DistinctDataBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/DistinctDataBag.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/DistinctDataBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/DistinctDataBag.java Fri Sep 28 18:30:22 2012
@@ -108,22 +108,10 @@ public class DistinctDataBag extends Def
                 mSize++;
             }
         }
+        markSpillableIfNecessary();
     }
 
     @Override
-    public void addAll(DataBag b) {
-        synchronized (mContents) {
-            Iterator<Tuple> i = b.iterator();
-            while (i.hasNext()) {
-                if (mContents.add(i.next())) {
-                    mSize++;
-                }
-            }
-        }
-    }
-
-
-    @Override
     public long spill() {
         // Make sure we have something to spill.  Don't create empty
         // files, as that will make a mess.

Modified: pig/trunk/src/org/apache/pig/data/InternalCachedBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/InternalCachedBag.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/InternalCachedBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/InternalCachedBag.java Fri Sep 28 18:30:22 2012
@@ -26,7 +26,6 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.NoSuchElementException;
 
@@ -116,22 +115,6 @@ public class InternalCachedBag extends S
         numTuplesSpilled = 0;
     }
 
-    @Override
-    public void addAll(DataBag b) {
-    	Iterator<Tuple> iter = b.iterator();
-    	while(iter.hasNext()) {
-    		add(iter.next());
-    	}
-    }
-
-    @Override
-    public void addAll(Collection<Tuple> c) {
-    	Iterator<Tuple> iter = c.iterator();
-    	while(iter.hasNext()) {
-    		add(iter.next());
-    	}
-    }
-    
     private void addDone() {
         if(out != null) {
             try {
@@ -145,6 +128,7 @@ public class InternalCachedBag extends S
         if(numTuplesSpilled > 0)
             updateSpillRecCounter();
         addDone = true;
+        markSpillableIfNecessary();
     }
 
     @Override

Modified: pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/InternalDistinctBag.java Fri Sep 28 18:30:22 2012
@@ -26,7 +26,6 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -150,26 +149,9 @@ public class InternalDistinctBag extends
                 memLimit.addNewObjSize(t.getMemorySize());
             }          
         }    	
+        markSpillableIfNecessary();
     }
 
-    @Override
-    public void addAll(DataBag b) {
-    	Iterator<Tuple> iter = b.iterator();
-    	while(iter.hasNext()) {
-    		add(iter.next());
-    	}
-    }
-
-    @Override
-    public void addAll(Collection<Tuple> c) {
-    	Iterator<Tuple> iter = c.iterator();
-    	while(iter.hasNext()) {
-    		add(iter.next());
-    	}
-    }
-    
-    
-   
     /**
      * An iterator that handles getting the next tuple from the bag.
      * Data can be stored in a combination of in memory and on disk.  

Modified: pig/trunk/src/org/apache/pig/data/InternalSortedBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/InternalSortedBag.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/InternalSortedBag.java (original)
+++ pig/trunk/src/org/apache/pig/data/InternalSortedBag.java Fri Sep 28 18:30:22 2012
@@ -26,7 +26,6 @@ import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Iterator;
@@ -36,7 +35,6 @@ import java.util.PriorityQueue;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce;
 
 
 /**
@@ -66,15 +64,18 @@ public class InternalSortedBag extends S
     private transient boolean mReadStarted = false;
 
     static private class DefaultComparator implements Comparator<Tuple> {
+        @Override
         @SuppressWarnings("unchecked")
 		public int compare(Tuple t1, Tuple t2) {
             return t1.compareTo(t2);
         }
 
+        @Override
         public boolean equals(Object o) {
         	return (o == this);
         }
 
+        @Override
         public int hashCode() {
             return 42; 
         }
@@ -107,6 +108,7 @@ public class InternalSortedBag extends S
      
     }
     
+    @Override
     public void add(Tuple t) {
     	if(mReadStarted) {
             throw new IllegalStateException("InternalSortedBag is closed for adding new tuples");
@@ -126,30 +128,20 @@ public class InternalSortedBag extends S
         }
                 
         mSize++;
+        markSpillableIfNecessary();
     }
     
-    public void addAll(DataBag b) {
-    	Iterator<Tuple> iter = b.iterator();
-    	while(iter.hasNext()) {
-    		add(iter.next());
-    	}
-    }
-
-    public void addAll(Collection<Tuple> c) {
-    	Iterator<Tuple> iter = c.iterator();
-    	while(iter.hasNext()) {
-    		add(iter.next());
-    	}
-    }    
-
+    @Override
     public boolean isSorted() {
         return true;
     }
     
+    @Override
     public boolean isDistinct() {
         return false;
     }
     
+    @Override
     public Iterator<Tuple> iterator() {
         return new SortedDataBagIterator();
     }
@@ -170,10 +162,12 @@ public class InternalSortedBag extends S
             public Tuple tuple;
             public int fileNum;
 
+            @Override
             public int compareTo(PQContainer other) {
                 return mComp.compare(tuple, other.tuple);
             }
             
+            @Override
             public boolean equals(Object obj) {
             	if (obj instanceof PQContainer) {
             		return compareTo((PQContainer)obj) == 0;
@@ -182,6 +176,7 @@ public class InternalSortedBag extends S
             	return false;
             }
             
+            @Override
             public int hashCode() {
             	return tuple.hashCode();
             }
@@ -205,12 +200,14 @@ public class InternalSortedBag extends S
             }            
         }
 
+        @Override
         public boolean hasNext() { 
             // See if we can find a tuple.  If so, buffer it.
             mBuf = next();
             return mBuf != null;
         }
 
+        @Override
         public Tuple next() {
             // This will report progress every 1024 times through next.
             // This should be much faster than using mod.
@@ -235,6 +232,7 @@ public class InternalSortedBag extends S
         /**
          * Not implemented.
          */
+        @Override
         public void remove() {}
 
         private Tuple readFromPriorityQ() {
@@ -436,6 +434,7 @@ public class InternalSortedBag extends S
         }
     }
 
+    @Override
     public long spill(){
         return proactive_spill(mComp);
     }

Modified: pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java (original)
+++ pig/trunk/src/org/apache/pig/impl/util/SpillableMemoryManager.java Fri Sep 28 18:30:22
2012
@@ -78,7 +78,9 @@ public class SpillableMemoryManager impl
     // log notification on collection threshold exceeded only the first time
     private boolean firstCollectionThreshExceededLogged = false;
     
-    public SpillableMemoryManager() {
+    private static SpillableMemoryManager manager;
+
+    private SpillableMemoryManager() {
         ((NotificationEmitter)ManagementFactory.getMemoryMXBean()).addNotificationListener(this,
null, null);
         List<MemoryPoolMXBean> mpbeans = ManagementFactory.getMemoryPoolMXBeans();
         MemoryPoolMXBean biggestHeap = null;
@@ -120,6 +122,13 @@ public class SpillableMemoryManager impl
         biggestHeap.setUsageThreshold((long)(biggestSize * memoryThresholdFraction));
     }
     
+    public static SpillableMemoryManager getInstance() {
+        if (manager == null) {
+            manager = new SpillableMemoryManager();
+        }
+        return manager;
+    }
+
     public static void configure(Properties properties) {
         
         try {
@@ -136,6 +145,7 @@ public class SpillableMemoryManager impl
         }
     }
     
+    @Override
     public void handleNotification(Notification n, Object o) {
         CompositeData cd = (CompositeData) n.getUserData();
         MemoryNotificationInfo info = MemoryNotificationInfo.from(cd);
@@ -195,6 +205,7 @@ public class SpillableMemoryManager impl
                  * Also between the time we sort and we use these spillables, they
                  * may actually change in size - so this is just best effort
                  */    
+                @Override
                 public int compare(WeakReference<Spillable> o1Ref, WeakReference<Spillable>
o2Ref) {
                     Spillable o1 = o1Ref.get();
                     Spillable o2 = o2Ref.get();

Modified: pig/trunk/test/org/apache/pig/test/TestDataBag.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestDataBag.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestDataBag.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestDataBag.java Fri Sep 28 18:30:22 2012
@@ -17,6 +17,8 @@
  */
 package org.apache.pig.test;
 
+import static org.junit.Assert.*;
+
 import java.util.*;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
@@ -24,20 +26,22 @@ import java.io.IOException;
 import java.io.PipedInputStream;
 import java.io.PipedOutputStream;
 
-import org.junit.Test;
+
 import org.apache.pig.data.*;
 import org.apache.pig.impl.util.Spillable;
+import org.junit.After;
+import org.junit.Test;
 
 
 /**
  * This class will exercise the basic Pig data model and members. It tests for proper behavior
in
- * assigment and comparision, as well as function application.
+ * assignment and comparison, as well as function application.
  * 
  * @author dnm
  */
-public class TestDataBag extends junit.framework.TestCase {
+public class TestDataBag  {
 
-    private Random rand = new Random();
+    private Random rand = new Random(111);
 
     private class TestMemoryManager {
         ArrayList<Spillable> mManagedObjects = new ArrayList<Spillable>();
@@ -47,8 +51,9 @@ public class TestDataBag extends junit.f
         }
 
         public void forceSpill() throws IOException {
-            Iterator<Spillable> i = mManagedObjects.iterator();
-            while (i.hasNext()) i.next().spill();
+            for (Spillable spillable : mManagedObjects) {
+                spillable.spill();
+            }
         }
     }
 
@@ -80,8 +85,8 @@ public class TestDataBag extends junit.f
         }
     }
 
-    @Override
-    protected void tearDown() throws Exception {
+    @After
+    public void tearDown() throws Exception {
     	BagFactory.resetSelf();
         System.clearProperty("pig.data.bag.factory.name");
         System.clearProperty("pig.data.bag.factory.jar");
@@ -586,7 +591,7 @@ public class TestDataBag extends junit.f
             mgr.forceSpill();
         }
         
-        assertEquals("Size of distinct data bag is incorrect", b.size(), rightAnswer.size());
+       assertEquals("Size of distinct data bag is incorrect", rightAnswer.size(), b.size());
 
         // Read tuples back, hopefully they come out in the same order.
         Iterator<Tuple> bIter = b.iterator();

Modified: pig/trunk/test/org/apache/pig/test/TestDataBagAccess.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestDataBagAccess.java?rev=1391584&r1=1391583&r2=1391584&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestDataBagAccess.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestDataBagAccess.java Fri Sep 28 18:30:22 2012
@@ -18,6 +18,8 @@
 
 package org.apache.pig.test;
 
+import static org.junit.Assert.*;
+
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -41,20 +43,14 @@ import org.apache.pig.impl.logicalLayer.
 import org.apache.pig.impl.util.MultiMap;
 import org.junit.Before;
 import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-
-import junit.framework.TestCase;
 
 /**
  *
  */
-@RunWith(JUnit4.class)
-public class TestDataBagAccess extends TestCase {
+public class TestDataBagAccess {
     private PigServer pigServer;
 
     @Before
-    @Override
     public void setUp() throws Exception{
         pigServer = new PigServer(ExecType.LOCAL, new Properties());
     }
@@ -144,11 +140,11 @@ public class TestDataBagAccess extends T
         // (p1-t1-e2,p1-t2-e1)
         // (p2-t1-e2,p2-t2-e1)
         Tuple t = it.next();
-        assertEquals("p1-t1-e2", (String)t.get(0));
-        assertEquals("p1-t2-e1", (String)t.get(1));
+        assertEquals("p1-t1-e2", t.get(0));
+        assertEquals("p1-t2-e1", t.get(1));
         t = it.next();
-        assertEquals("p2-t1-e2", (String)t.get(0));
-        assertEquals("p2-t2-e1", (String)t.get(1));
+        assertEquals("p2-t1-e2", t.get(0));
+        assertEquals("p2-t2-e1", t.get(1));
         assertFalse(it.hasNext());
     }
 



Mime
View raw message