HADOOP-212. Permit alteration of the file block size in DFS. Contributed by Owen.

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

CHANGES.txt

@@ -29,6 +29,12 @@ Trunk (unreleased)
  7. HADOOP-219. Fix a NullPointerException when handling a checksum
     exception under SequenceFile.Sorter.sort().  (cutting & stack)
 
+ 8. HADOOP-212. Permit alteration of the file block size in DFS.  The
+    default block size for new files may now be specified in the
+    configuration with the dfs.block.size property.  The block size
+    may also be specified when files are opened.
+    (omalley via cutting)
+
 
 Release 0.2.1 - 2006-05-12
 

conf/hadoop-default.xml

@@ -134,6 +134,12 @@ creations/deletions), or "all".</description>
   </description>
 </property>
 
+<property>
+  <name>dfs.block.size</name>
+  <value>67108864</value>
+  <description>The default block size for new files.</description>
+</property>
+
 <property>
   <name>dfs.df.interval</name>
   <value>3000</value>

src/java/org/apache/hadoop/dfs/Block.java

@@ -110,10 +110,9 @@ class Block implements Writable, Comparable {
     public void readFields(DataInput in) throws IOException {
         this.blkid = in.readLong();
         this.len = in.readLong();
-        if( len < 0 || len > FSConstants.BLOCK_SIZE )
-          throw new IOException("Unexpected block size: " + len 
-                                + ". System block size = " 
-                                + FSConstants.BLOCK_SIZE + ".");
+        if( len < 0 ) {
+          throw new IOException("Unexpected block size: " + len);
+        }
     }
 
     /////////////////////////////////////

src/java/org/apache/hadoop/dfs/ClientProtocol.java

@@ -55,7 +55,8 @@ interface ClientProtocol {
                                 String clientName, 
                                 String clientMachine, 
                                 boolean overwrite, 
-                                short replication
+                                short replication,
+                                long blockSize
                               ) throws IOException;
 
     /**
@@ -231,4 +232,12 @@ interface ClientProtocol {
      * One DatanodeInfo object is returned for each DataNode.
      */
     public DatanodeInfo[] getDatanodeReport() throws IOException;
+
+    /**
+     * Get the block size for the given file.
+     * @param filename The name of the file
+     * @return The number of bytes in each block
+     * @throws IOException
+     */
+    public long getBlockSize(String filename) throws IOException;
 }

src/java/org/apache/hadoop/dfs/DFSClient.java

@@ -41,6 +41,7 @@ import java.util.logging.*;
 class DFSClient implements FSConstants {
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.fs.DFSClient");
     static int MAX_BLOCK_ACQUIRE_FAILURES = 3;
+    private static final long DEFAULT_BLOCK_SIZE = 64 * 1024 * 1024;
     ClientProtocol namenode;
     String localName;
     boolean running = true;
@@ -48,6 +49,8 @@ class DFSClient implements FSConstants {
     String clientName;
     Daemon leaseChecker;
     private Configuration conf;
+    private long defaultBlockSize;
+    private short defaultReplication;
     
     // required for unknown reason to make WritableFactories work distributed
     static { new DFSFileInfo(); }
@@ -110,6 +113,8 @@ class DFSClient implements FSConstants {
         } else {
             this.clientName = "DFSClient_" + r.nextInt();
         }
+        defaultBlockSize = conf.getLong("dfs.block.size", DEFAULT_BLOCK_SIZE);
+        defaultReplication = (short) conf.getInt("dfs.replication", 3);
         this.leaseChecker = new Daemon(new LeaseChecker());
         this.leaseChecker.start();
     }
@@ -150,6 +155,37 @@ class DFSClient implements FSConstants {
       }
     }
 
+    /**
+     * Get the default block size for this cluster
+     * @return the default block size in bytes
+     */
+    public long getDefaultBlockSize() {
+      return defaultBlockSize;
+    }
+    
+    public long getBlockSize(Path f) throws IOException {
+      // if we already know the answer, use it.
+      if (f instanceof DfsPath) {
+        return ((DfsPath) f).getBlockSize();
+      }
+      int retries = 4;
+      while (true) {
+        try {
+          return namenode.getBlockSize(f.toString());
+        } catch (IOException ie) {
+          LOG.info("Problem getting block size: " + 
+                   StringUtils.stringifyException(ie));
+          if (--retries == 0) {
+            throw ie;
+          }
+        }
+      }
+    }
+    
+    public short getDefaultReplication() {
+      return defaultReplication;
+    }
+    
     /**
      * Get hints about the location of the indicated block(s).  The
      * array returned is as long as there are blocks in the indicated
@@ -182,7 +218,7 @@ class DFSClient implements FSConstants {
     public FSOutputStream create( UTF8 src, 
                                   boolean overwrite
                                 ) throws IOException {
-      return create( src, overwrite, (short)conf.getInt("dfs.replication", 3));
+      return create( src, overwrite, defaultReplication, defaultBlockSize);
     }
     
     /**
@@ -197,10 +233,12 @@ class DFSClient implements FSConstants {
      */
     public FSOutputStream create( UTF8 src, 
                                   boolean overwrite, 
-                                  short replication
+                                  short replication,
+                                  long blockSize
                                 ) throws IOException {
       checkOpen();
-      FSOutputStream result = new DFSOutputStream(src, overwrite, replication);
+      FSOutputStream result = new DFSOutputStream(src, overwrite, 
+                                                  replication, blockSize);
       synchronized (pendingCreates) {
         pendingCreates.put(src.toString(), result);
       }
@@ -672,15 +710,19 @@ class DFSClient implements FSConstants {
         private long filePos = 0;
         private int bytesWrittenToBlock = 0;
         private String datanodeName;
+        private long blockSize;
 
         /**
          * Create a new output stream to the given DataNode.
          */
-        public DFSOutputStream(UTF8 src, boolean overwrite, short replication) throws IOException {
+        public DFSOutputStream(UTF8 src, boolean overwrite, 
+                               short replication, long blockSize
+                               ) throws IOException {
             this.src = src;
             this.overwrite = overwrite;
             this.replication = replication;
             this.backupFile = newBackupFile();
+            this.blockSize = blockSize;
             this.backupStream = new FileOutputStream(backupFile);
         }
 
@@ -766,7 +808,7 @@ class DFSClient implements FSConstants {
             while (true) {
               try {
                 return namenode.create(src.toString(), clientName.toString(),
-                    localName, overwrite, replication);
+                    localName, overwrite, replication, blockSize);
               } catch (RemoteException e) {
                 if (--retries == 0 || 
                     "org.apache.hadoop.dfs.NameNode.AlreadyBeingCreatedException".
@@ -835,7 +877,7 @@ class DFSClient implements FSConstants {
                 throw new IOException("Stream closed");
             }
 
-            if ((bytesWrittenToBlock + pos == BLOCK_SIZE) ||
+            if ((bytesWrittenToBlock + pos == blockSize) ||
                 (pos >= BUFFER_SIZE)) {
                 flush();
             }
@@ -861,7 +903,7 @@ class DFSClient implements FSConstants {
               len -= toWrite;
               filePos += toWrite;
 
-              if ((bytesWrittenToBlock + pos >= BLOCK_SIZE) ||
+              if ((bytesWrittenToBlock + pos >= blockSize) ||
                   (pos == BUFFER_SIZE)) {
                 flush();
               }
@@ -877,10 +919,10 @@ class DFSClient implements FSConstants {
                 throw new IOException("Stream closed");
             }
 
-            if (bytesWrittenToBlock + pos >= BLOCK_SIZE) {
-                flushData(BLOCK_SIZE - bytesWrittenToBlock);
+            if (bytesWrittenToBlock + pos >= blockSize) {
+                flushData((int) blockSize - bytesWrittenToBlock);
             }
-            if (bytesWrittenToBlock == BLOCK_SIZE) {
+            if (bytesWrittenToBlock == blockSize) {
                 endBlock();
             }
             flushData(pos);

src/java/org/apache/hadoop/dfs/DFSFileInfo.java

@@ -40,6 +40,7 @@ class DFSFileInfo implements Writable {
     long contentsLen;
     boolean isDir;
     short blockReplication;
+    long blockSize;
 
     /**
      */
@@ -58,6 +59,7 @@ class DFSFileInfo implements Writable {
         } else 
           this.len = this.contentsLen = node.computeFileLength();
         this.blockReplication = node.getReplication();
+        blockSize = node.getBlockSize();
     }
 
     /**
@@ -102,6 +104,14 @@ class DFSFileInfo implements Writable {
       return this.blockReplication;
     }
 
+    /**
+     * Get the block size of the file.
+     * @return the number of bytes
+     */
+    public long getBlockSize() {
+      return blockSize;
+    }
+    
     //////////////////////////////////////////////////
     // Writable
     //////////////////////////////////////////////////
@@ -111,6 +121,7 @@ class DFSFileInfo implements Writable {
         out.writeLong(contentsLen);
         out.writeBoolean(isDir);
         out.writeShort(blockReplication);
+        out.writeLong(blockSize);
     }
 
     public void readFields(DataInput in) throws IOException {
@@ -120,6 +131,7 @@ class DFSFileInfo implements Writable {
         this.contentsLen = in.readLong();
         this.isDir = in.readBoolean();
         this.blockReplication = in.readShort();
+        blockSize = in.readLong();
     }
 }
 

src/java/org/apache/hadoop/dfs/DFSck.java

@@ -106,7 +106,7 @@ public class DFSck {
   public Result fsck(String path) throws Exception {
     DFSFileInfo[] files = dfs.listPaths(new UTF8(path));
     Result res = new Result();
-    res.setReplication(conf.getInt("dfs.replication", 3));
+    res.setReplication(dfs.getDefaultReplication());
     for (int i = 0; i < files.length; i++) {
       check(files[i], res);
     }

src/java/org/apache/hadoop/dfs/DfsPath.java

@@ -18,7 +18,11 @@ package org.apache.hadoop.dfs;
 import org.apache.hadoop.fs.Path;
 
 
-/** DfsPath is a Path that's been annotated with some extra information. */
+/** 
+ * DfsPath is a Path that's been annotated with some extra information.
+ * The point of this class is to pass back the "common" metadata about 
+ * a file with the names in a directory listing to make accesses faster.
+ */
 class DfsPath extends Path {
     DFSFileInfo info;
 
@@ -42,4 +46,7 @@ class DfsPath extends Path {
     public short getReplication() {
       return info.getReplication();
     }
+    public long getBlockSize() {
+      return info.getBlockSize();
+    }
 }

src/java/org/apache/hadoop/dfs/DistributedFileSystem.java

@@ -54,6 +54,18 @@ public class DistributedFileSystem extends FileSystem {
       return workingDir;
     }
     
+    public long getDefaultBlockSize() {
+      return dfs.getDefaultBlockSize();
+    }
+    
+    public long getBlockSize(Path f) throws IOException {
+      return dfs.getBlockSize(f);
+    }
+    
+    public short getDefaultReplication() {
+      return dfs.getDefaultReplication();
+    }
+    
     private Path makeAbsolute(Path f) {
       if (f.isAbsolute()) {
         return f;
@@ -78,9 +90,10 @@ public class DistributedFileSystem extends FileSystem {
       return dfs.open(getPath(f));
     }
 
-    public FSOutputStream createRaw(Path f, boolean overwrite, short replication)
+    public FSOutputStream createRaw(Path f, boolean overwrite, 
+                                    short replication, long blockSize)
       throws IOException {
-      return dfs.create(getPath(f), overwrite, replication);
+      return dfs.create(getPath(f), overwrite, replication, blockSize);
     }
 
     public boolean setReplicationRaw( Path src, 
@@ -205,10 +218,6 @@ public class DistributedFileSystem extends FileSystem {
       // no data is lost. a task may fail, but on retry it should succeed.
     }
 
-    public long getBlockSize() {
-      return FSConstants.BLOCK_SIZE;
-    }
-
     /** Return the total raw capacity of the filesystem, disregarding
      * replication .*/
     public long getRawCapacity() throws IOException{

src/java/org/apache/hadoop/dfs/FSConstants.java

@@ -23,7 +23,6 @@ import org.apache.hadoop.conf.Configuration;
  * @author Mike Cafarella
  ************************************/
 interface FSConstants {
-    public static int BLOCK_SIZE = 32 * 1000 * 1000;
     public static int MIN_BLOCKS_FOR_WRITE = 5;
 
     public static final long WRITE_COMPLETE = 0xcafae11a;

src/java/org/apache/hadoop/dfs/FSDataset.java

@@ -263,6 +263,7 @@ class FSDataset implements FSConstants {
         if (isValidBlock(b)) {
             throw new IOException("Block " + b + " is valid, and cannot be written to.");
         }
+        long blockSize = b.getNumBytes();
 
         //
         // Serialize access to /tmp, and check if file already there.
@@ -279,7 +280,7 @@ class FSDataset implements FSConstants {
             //
             // Check if we have too little space
             //
-            if (getRemaining() < BLOCK_SIZE) {
+            if (getRemaining() < blockSize) {
                 throw new IOException("Insufficient space for an additional block");
             }
 
@@ -288,7 +289,7 @@ class FSDataset implements FSConstants {
             // 'reserved' size, & create file
             //
             ongoingCreates.add(b);
-            reserved += BLOCK_SIZE;
+            reserved += blockSize;
             f = getTmpFile(b);
 	    try {
 		if (f.exists()) {
@@ -304,7 +305,7 @@ class FSDataset implements FSConstants {
 	    } catch (IOException ie) {
                 System.out.println("Exception!  " + ie);
 		ongoingCreates.remove(b);		
-		reserved -= BLOCK_SIZE;
+		reserved -= blockSize;
                 throw ie;
 	    }
         }
@@ -358,7 +359,7 @@ class FSDataset implements FSConstants {
             if (! ongoingCreates.remove(b)) {
                 throw new IOException("Tried to finalize block " + b + ", but could not find it in ongoingCreates after file-move!");
             } 
-            reserved -= BLOCK_SIZE;
+            reserved -= b.getNumBytes();
         }
     }
 

src/java/org/apache/hadoop/dfs/FSDirectory.java

@@ -240,6 +240,18 @@ class FSDirectory implements FSConstants {
             return total;
         }
 
+        /**
+         * Get the block size of the first block
+         * @return the number of bytes
+         */
+        public long getBlockSize() {
+          if (blocks == null || blocks.length == 0) {
+            return 0;
+          } else {
+            return blocks[0].getNumBytes();
+          }
+        }
+        
         /**
          */
         void listContents(Vector v) {
@@ -747,7 +759,27 @@ class FSDirectory implements FSConstants {
       }
       return fileBlocks;
     }
-                                 
+
+    /**
+     * Get the blocksize of a file
+     * @param filename the filename
+     * @return the number of bytes in the first block
+     * @throws IOException if it is a directory or does not exist.
+     */
+    public long getBlockSize(String filename) throws IOException {
+      synchronized (rootDir) {
+        INode fileNode = rootDir.getNode(filename);
+        if (fileNode == null) {
+          throw new IOException("Unknown file: " + filename);
+        }
+        if (fileNode.isDir()) {
+          throw new IOException("Getting block size of a directory: " + 
+                                filename);
+        }
+        return fileNode.getBlockSize();
+      }
+    }
+    
     /**
      * Remove the file from management, return blocks
      */

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -37,10 +37,6 @@ import java.util.logging.*;
  ***************************************************/
 class FSNamesystem implements FSConstants {
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.fs.FSNamesystem");
-    static {
-      // for debugging the pending Creates problems
-      LOG.setLevel(Level.FINE);
-    }
 
     //
     // Stores the correct file name hierarchy
@@ -276,6 +272,10 @@ class FSNamesystem implements FSConstants {
       return true;
     }
     
+    public long getBlockSize(String filename) throws IOException {
+      return dir.getBlockSize(filename);
+    }
+    
     /**
      * Check whether the replication parameter is within the range
      * determined by system configuration.
@@ -312,7 +312,8 @@ class FSNamesystem implements FSConstants {
                                             UTF8 holder, 
                                             UTF8 clientMachine, 
                                             boolean overwrite,
-                                            short replication 
+                                            short replication,
+                                            long blockSize
                                           ) throws IOException {
       NameNode.stateChangeLog.fine("DIR* NameSystem.startFile: file "
             +src+" for "+holder+" at "+clientMachine);
@@ -340,7 +341,8 @@ class FSNamesystem implements FSConstants {
         }
 
         // Get the array of replication targets 
-        DatanodeInfo targets[] = chooseTargets(replication, null, clientMachine);
+        DatanodeInfo targets[] = chooseTargets(replication, null, 
+                                               clientMachine, blockSize);
         if (targets.length < this.minReplication) {
             throw new IOException("failed to create file "+src
                     +" on client " + clientMachine
@@ -351,6 +353,7 @@ class FSNamesystem implements FSConstants {
         // Reserve space for this pending file
         pendingCreates.put(src, 
                            new FileUnderConstruction(replication, 
+                                                     blockSize,
                                                      holder,
                                                      clientMachine));
         NameNode.stateChangeLog.finer( "DIR* NameSystem.startFile: "
@@ -421,7 +424,7 @@ class FSNamesystem implements FSConstants {
         
         // Get the array of replication targets 
         DatanodeInfo targets[] = chooseTargets(pendingFile.getReplication(), 
-            null, pendingFile.getClientMachine());
+            null, pendingFile.getClientMachine(), pendingFile.getBlockSize());
         if (targets.length < this.minReplication) {
           throw new IOException("File " + src + " could only be replicated to " +
                                 targets.length + " nodes, instead of " +
@@ -1400,6 +1403,7 @@ class FSNamesystem implements FSConstants {
                     }
 
                     Block block = (Block) it.next();
+                    long blockSize = block.getNumBytes();
                     FSDirectory.INode fileINode = dir.getFileByBlock(block);
                     if( fileINode == null ) { // block does not belong to any file
                         it.remove();
@@ -1411,7 +1415,13 @@ class FSNamesystem implements FSConstants {
                         if (containingNodes.contains(srcNode) 
                               && ( excessBlocks == null 
                                || ! excessBlocks.contains(block))) {
-                            DatanodeInfo targets[] = chooseTargets(Math.min(fileINode.getReplication() - containingNodes.size(), this.maxReplicationStreams - xmitsInProgress), containingNodes, null);
+                            DatanodeInfo targets[] = 
+                              chooseTargets(Math.min(fileINode.getReplication() 
+                                                       - containingNodes.size(), 
+                                                     maxReplicationStreams
+                                                       - xmitsInProgress), 
+                                                     containingNodes, null, 
+                                                     blockSize);
                             if (targets.length > 0) {
                                 // Build items to return
                                 replicateBlocks.add(block);
@@ -1481,7 +1491,8 @@ class FSNamesystem implements FSConstants {
      * considered targets.
      * @return array of DatanodeInfo instances uses as targets.
      */
-    DatanodeInfo[] chooseTargets(int desiredReplicates, TreeSet forbiddenNodes, UTF8 clientMachine) {
+    DatanodeInfo[] chooseTargets(int desiredReplicates, TreeSet forbiddenNodes,
+                                 UTF8 clientMachine, long blockSize) {
         if (desiredReplicates > datanodeMap.size()) {
           LOG.warning("Replication requested of "+desiredReplicates
                       +" is larger than cluster size ("+datanodeMap.size()
@@ -1493,7 +1504,8 @@ class FSNamesystem implements FSConstants {
         Vector targets = new Vector();
 
         for (int i = 0; i < desiredReplicates; i++) {
-            DatanodeInfo target = chooseTarget(forbiddenNodes, alreadyChosen, clientMachine);
+            DatanodeInfo target = chooseTarget(forbiddenNodes, alreadyChosen, 
+                                               clientMachine, blockSize);
             if (target == null)
               break; // calling chooseTarget again won't help
             targets.add(target);
@@ -1514,7 +1526,8 @@ class FSNamesystem implements FSConstants {
      * @return DatanodeInfo instance to use or null if something went wrong
      * (a log message is emitted if null is returned).
      */
-    DatanodeInfo chooseTarget(TreeSet forbidden1, TreeSet forbidden2, UTF8 clientMachine) {
+    DatanodeInfo chooseTarget(TreeSet forbidden1, TreeSet forbidden2, 
+                              UTF8 clientMachine, long blockSize) {
         //
         // Check if there are any available targets at all
         //
@@ -1565,7 +1578,7 @@ class FSNamesystem implements FSConstants {
                 for (Iterator it = targetList.iterator(); it.hasNext(); ) {
                     DatanodeInfo node = (DatanodeInfo) it.next();
                     if (clientMachine.equals(node.getHost())) {
-                        if (node.getRemaining() > BLOCK_SIZE * MIN_BLOCKS_FOR_WRITE) {
+                        if (node.getRemaining() > blockSize * MIN_BLOCKS_FOR_WRITE) {
                             return node;
                         }
                     }
@@ -1577,7 +1590,7 @@ class FSNamesystem implements FSConstants {
             //
             for (Iterator it = targetList.iterator(); it.hasNext(); ) {
                 DatanodeInfo node = (DatanodeInfo) it.next();
-                if (node.getRemaining() > BLOCK_SIZE * MIN_BLOCKS_FOR_WRITE) {
+                if (node.getRemaining() > blockSize * MIN_BLOCKS_FOR_WRITE) {
                     return node;
                 }
             }
@@ -1589,7 +1602,7 @@ class FSNamesystem implements FSConstants {
             //
             for (Iterator it = targetList.iterator(); it.hasNext(); ) {
                 DatanodeInfo node = (DatanodeInfo) it.next();
-                if (node.getRemaining() > BLOCK_SIZE) {
+                if (node.getRemaining() > blockSize) {
                     return node;
                 }
             }
@@ -1615,14 +1628,17 @@ class FSNamesystem implements FSConstants {
      */
     private class FileUnderConstruction {
       private short blockReplication; // file replication
+      private long blockSize;
       private Vector blocks;
       private UTF8 clientName;         // lease holder
       private UTF8 clientMachine;
       
       FileUnderConstruction(short replication,
+                            long blockSize,
                             UTF8 clientName,
                             UTF8 clientMachine) throws IOException {
         this.blockReplication = replication;
+        this.blockSize = blockSize;
         this.blocks = new Vector();
         this.clientName = clientName;
         this.clientMachine = clientMachine;
@@ -1632,6 +1648,10 @@ class FSNamesystem implements FSConstants {
         return this.blockReplication;
       }
       
+      public long getBlockSize() {
+        return blockSize;
+      }
+      
       public Vector getBlocks() {
         return blocks;
       }

src/java/org/apache/hadoop/dfs/NameNode.java

@@ -181,7 +181,8 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
                                String clientName, 
                                String clientMachine, 
                                boolean overwrite,
-                               short replication
+                               short replication,
+                               long blockSize
     ) throws IOException {
        stateChangeLog.fine("*DIR* NameNode.create: file "
             +src+" for "+clientName+" at "+clientMachine);
@@ -189,7 +190,8 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
                                                 new UTF8(clientName), 
                                                 new UTF8(clientMachine), 
                                                 overwrite,
-                                                replication);
+                                                replication,
+                                                blockSize);
         Block b = (Block) results[0];
         DatanodeInfo targets[] = (DatanodeInfo[]) results[1];
         return new LocatedBlock(b, targets);
@@ -279,6 +281,11 @@ public class NameNode implements ClientProtocol, DatanodeProtocol, FSConstants {
             return results;
         }
     }
+    
+    public long getBlockSize(String filename) throws IOException {
+      return namesystem.getBlockSize(filename);
+    }
+    
     /**
      */
     public boolean rename(String src, String dst) throws IOException {

src/java/org/apache/hadoop/fs/FSDataOutputStream.java

@@ -38,12 +38,14 @@ public class FSDataOutputStream extends DataOutputStream {
                   Path file, 
                   boolean overwrite, 
                   short replication,
+                  long blockSize,
                   Configuration conf)
       throws IOException {
-      super(fs.createRaw(file, overwrite, replication));
+      super(fs.createRaw(file, overwrite, replication, blockSize));
       this.bytesPerSum = conf.getInt("io.bytes.per.checksum", 512);
       this.sums = new FSDataOutputStream(
-            fs.createRaw(FileSystem.getChecksumFile(file), true, replication), 
+            fs.createRaw(FileSystem.getChecksumFile(file), true, 
+                         replication, blockSize), 
             conf);
       sums.write(CHECKSUM_VERSION, 0, CHECKSUM_VERSION.length);
       sums.writeInt(this.bytesPerSum);
@@ -127,11 +129,11 @@ public class FSDataOutputStream extends DataOutputStream {
 
   public FSDataOutputStream(FileSystem fs, Path file,
                             boolean overwrite, Configuration conf,
-                            int bufferSize, short replication )
+                            int bufferSize, short replication, long blockSize )
   throws IOException {
     super(new Buffer(
             new PositionCache(
-                new Summer(fs, file, overwrite, replication, conf)), 
+                new Summer(fs, file, overwrite, replication, blockSize, conf)), 
             bufferSize));
   }
 

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -175,7 +175,8 @@ public abstract class FileSystem extends Configured {
     public FSDataOutputStream create(Path f) throws IOException {
       return create(f, true, 
                     getConf().getInt("io.file.buffer.size", 4096),
-                    (short)getConf().getInt("dfs.replication", 3));
+                    getDefaultReplication(),
+                    getDefaultBlockSize());
     }
 
     /**
@@ -186,7 +187,8 @@ public abstract class FileSystem extends Configured {
       throws IOException {
       return create(f, true, 
                     getConf().getInt("io.file.buffer.size", 4096),
-                    replication);
+                    replication,
+                    getDefaultBlockSize());
     }
 
     /**
@@ -201,7 +203,8 @@ public abstract class FileSystem extends Configured {
                                       int bufferSize
                                     ) throws IOException {
       return create( f, overwrite, bufferSize, 
-                    (short)getConf().getInt("dfs.replication", 3));
+                     getDefaultReplication(),
+                     getDefaultBlockSize());
     }
     
     /**
@@ -215,10 +218,11 @@ public abstract class FileSystem extends Configured {
     public FSDataOutputStream create( Path f, 
                                       boolean overwrite,
                                       int bufferSize,
-                                      short replication
+                                      short replication,
+                                      long blockSize
                                     ) throws IOException {
       return new FSDataOutputStream(this, f, overwrite, getConf(), 
-                                    bufferSize, replication );
+                                    bufferSize, replication, blockSize );
     }
 
     /** Opens an OutputStream at the indicated Path.
@@ -227,7 +231,9 @@ public abstract class FileSystem extends Configured {
      *   the file will be overwritten, and if false an error will be thrown.
      * @param replication required block replication for the file. 
      */
-    public abstract FSOutputStream createRaw(Path f, boolean overwrite, short replication)
+    public abstract FSOutputStream createRaw(Path f, boolean overwrite, 
+                                             short replication,
+                                             long blockSize)
       throws IOException;
 
     /** @deprecated Call {@link #createNewFile(Path)} instead. */
@@ -547,8 +553,20 @@ public abstract class FileSystem extends Configured {
                                                long start, long length,
                                                int crc);
 
+    /**
+     * Get the size for a particular file.
+     * @param f the filename
+     * @return the number of bytes in a block
+     */
+    public abstract long getBlockSize(Path f) throws IOException;
+    
     /** Return the number of bytes that large input files should be optimally
      * be split into to minimize i/o time. */
-    public abstract long getBlockSize();
+    public abstract long getDefaultBlockSize();
+    
+    /**
+     * Get the default replication.
+     */
+    public abstract short getDefaultReplication();
 
 }

src/java/org/apache/hadoop/fs/LocalFileSystem.java

@@ -161,7 +161,8 @@ public class LocalFileSystem extends FileSystem {
       }
     }
 
-    public FSOutputStream createRaw(Path f, boolean overwrite, short replication)
+    public FSOutputStream createRaw(Path f, boolean overwrite, 
+                                    short replication, long blockSize)
       throws IOException {
         if (exists(f) && ! overwrite) {
             throw new IOException("File already exists:"+f);
@@ -356,10 +357,19 @@ public class LocalFileSystem extends FileSystem {
       }
     }
 
-    public long getBlockSize() {
+    public long getDefaultBlockSize() {
       // default to 32MB: large enough to minimize the impact of seeks
       return getConf().getLong("fs.local.block.size", 32 * 1024 * 1024);
     }
 
+    public long getBlockSize(Path filename) {
+      // local doesn't really do blocks, so just use the global number
+      return getDefaultBlockSize();
+    }
+    
+    public short getDefaultReplication() {
+      return 1;
+    }
+
 
 }

src/java/org/apache/hadoop/mapred/InputFormatBase.java

@@ -32,7 +32,7 @@ public abstract class InputFormatBase implements InputFormat {
   public static final Logger LOG =
     LogFormatter.getLogger("org.apache.hadoop.mapred.InputFormatBase");
 
-  private static final double SPLIT_SLOP = 0.1;   // 10% slop
+  private static final double SPLIT_SLOP = 1.1;   // 10% slop
 
   private long minSplitSize = 1;
 
@@ -117,34 +117,22 @@ public abstract class InputFormatBase implements InputFormat {
       totalSize += fs.getLength(files[i]);
     }
 
-    long bytesPerSplit = totalSize / numSplits;   // start w/ desired num splits
+    long goalSize = totalSize / numSplits;   // start w/ desired num splits
 
-    long fsBlockSize = fs.getBlockSize();
-    if (bytesPerSplit > fsBlockSize) {            // no larger than fs blocks
-      bytesPerSplit = fsBlockSize;
-    }
-
-    long configuredMinSplitSize = job.getLong("mapred.min.split.size", 0);
-    if( configuredMinSplitSize < minSplitSize )
-    	configuredMinSplitSize = minSplitSize;
-    if (bytesPerSplit < configuredMinSplitSize) { // no smaller than min size
-      bytesPerSplit = configuredMinSplitSize;
-    }
-
-    long maxPerSplit = bytesPerSplit + (long)(bytesPerSplit*SPLIT_SLOP);
-
-    //LOG.info("bytesPerSplit = " + bytesPerSplit);
-    //LOG.info("maxPerSplit = " + maxPerSplit);
+    long minSize = Math.max(job.getLong("mapred.min.split.size", 1),
+                            minSplitSize);
 
     ArrayList splits = new ArrayList(numSplits);  // generate splits
     for (int i = 0; i < files.length; i++) {
       Path file = files[i];
       long length = fs.getLength(file);
+      long blockSize = fs.getBlockSize(file);
+      long splitSize = computeSplitSize(goalSize, minSize, blockSize);
 
       long bytesRemaining = length;
-      while (bytesRemaining >= maxPerSplit) {
-        splits.add(new FileSplit(file, length-bytesRemaining, bytesPerSplit));
-        bytesRemaining -= bytesPerSplit;
+      while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
+        splits.add(new FileSplit(file, length-bytesRemaining, splitSize));
+        bytesRemaining -= splitSize;
       }
       
       if (bytesRemaining != 0) {
@@ -156,5 +144,9 @@ public abstract class InputFormatBase implements InputFormat {
     return (FileSplit[])splits.toArray(new FileSplit[splits.size()]);
   }
 
+  private static long computeSplitSize(long goalSize, long minSize,
+                                       long blockSize) {
+    return Math.max(minSize, Math.min(goalSize, blockSize));
+  }
 }
 

src/test/org/apache/hadoop/fs/DistributedFSCheck.java

@@ -104,7 +104,7 @@ public class DistributedFSCheck extends TestCase {
     if( ! fs.isDirectory(rootFile) ) {
       nrFiles++;
       // For a regular file generate <fName,offset> pairs
-      long blockSize = fs.getBlockSize();
+      long blockSize = fs.getDefaultBlockSize();
       long fileLength = fs.getLength( rootFile );
       for( long offset = 0; offset < fileLength; offset += blockSize )
         writer.append(new UTF8(rootFile.toString()), new LongWritable(offset));
@@ -136,7 +136,7 @@ public class DistributedFSCheck extends TestCase {
       in = new DataInputStream(fs.open(new Path(name)));
       long actualSize = 0;
       try {
-        long blockSize = fs.getBlockSize();
+        long blockSize = fs.getDefaultBlockSize();
         int curSize = bufferSize;
         for(  actualSize = 0; 
               curSize == bufferSize && actualSize < blockSize;