svn commit: r1680975 - /hama/trunk/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java

[edwardyoon]

Author: edwardyoon
Date: Fri May 22 01:33:45 2015
New Revision: 1680975

URL: http://svn.apache.org/r1680975
Log:
HAMA-949: revert the commit. see the comments

Modified:
    hama/trunk/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java

Modified: hama/trunk/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java
URL: 
http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java?rev=1680975&r1=1680974&r2=1680975&view=diff
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java 
(original)
+++ hama/trunk/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java Fri 
May 22 01:33:45 2015
@@ -174,22 +174,28 @@ public abstract class FileInputFormat<K,
   public InputSplit[] getSplits(BSPJob job, int numSplits) throws IOException {
     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
     long maxSize = getMaxSplitSize(job);
-    
+
     // generate splits
     List<InputSplit> splits = new ArrayList<InputSplit>();
     FileStatus[] files = listStatus(job);
-    
+
+    /*
+     * TODO: This does not consider data locality. When the numSplits
+     * (user-defined) is equal to or smaller than the number of DFS splits, we
+     * should assign multiple splits to a task.
+     */
+
     // take the short circuit path if we have already partitioned
-    if (numSplits == files.length) {
-      for (FileStatus file : files) {
-        if (file != null) {
-          splits.add(new FileSplit(file.getPath(), 0, file.getLen(),
-              new String[0]));
-        }
-      }
-      return splits.toArray(new FileSplit[splits.size()]);
-    }
-    
+    // if (numSplits == files.length) {
+    // for (FileStatus file : files) {
+    // if (file != null) {
+    // splits.add(new FileSplit(file.getPath(), 0, file.getLen(),
+    // new String[0]));
+    // }
+    // }
+    // return splits.toArray(new FileSplit[splits.size()]);
+    // }
+
     for (FileStatus file : files) {
       Path path = file.getPath();
       FileSystem fs = path.getFileSystem(job.getConfiguration());
@@ -387,7 +393,8 @@ public abstract class FileInputFormat<K,
    * Add a {@link Path} to the list of inputs for the BSP job.
    * 
    * @param conf The configuration of the job
-   * @param p {@link Path} to be addaiaied to the list of inputs for the BSP 
job.
+   * @param p {@link Path} to be addaiaied to the list of inputs for the BSP
+   *          job.
    */
   public static void addInputPath(BSPJob conf, Path p) {
     Path path = new Path(conf.getWorkingDirectory(), p);