commit 56a11c6cbc552455218b89966924a9abba80f387
author Edward J. Yoon <edwardyoon@apache.org> Fri May 22 01:33:45 2015 +0000
committer Edward J. Yoon <edwardyoon@apache.org> Fri May 22 01:33:45 2015 +0000
tree c2a8fcd07702752d0f70264939a3d22829511e2c
parent 5a648dab6c4c90bfc5cc8b945e8815c024b33e0d

HAMA-949: revert the commit. see the comments

git-svn-id: https://svn.apache.org/repos/asf/hama/trunk@1680975 13f79535-47bb-0310-9956-ffa450edef68

core/src/main/java/org/apache/hama/bsp/FileInputFormat.java

diff --git a/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java b/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java
index 92f0cc2..eb57076 100644
--- a/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java
+++ b/core/src/main/java/org/apache/hama/bsp/FileInputFormat.java
@@ -174,22 +174,28 @@
   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 @@
    * 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);