diff --git a/CHANGES.txt b/CHANGES.txt
index bd94fc1..3dc4b5a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -155,6 +155,9 @@
     MAPREDUCE-2051. Contribute a fair scheduler preemption system test.
     (Todd Lipcon via tomwhite)
 
+    MAPREDUCE-1892. RaidNode can allow layered policies more efficiently.
+    (Ramkumar Vadali via schen)
+
   OPTIMIZATIONS
 
     MAPREDUCE-1354. Enhancements to JobTracker for better performance and
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/DirectoryTraversal.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/DirectoryTraversal.java
index 180d71f..5ced753 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/DirectoryTraversal.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/DirectoryTraversal.java
@@ -31,6 +31,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
 
 /**
  * Implements depth-first traversal using a Stack object. The traversal
@@ -46,6 +47,16 @@
   private Stack<Node> stack = new Stack<Node>();
 
   /**
+   * A FileFilter object can be used to choose files during directory traversal.
+   */
+  public interface FileFilter {
+    /**
+     * @return a boolean value indicating if the file passes the filter.
+     */
+    boolean check(FileStatus f) throws IOException;
+  }
+
+  /**
    * Represents a directory node in directory traversal.
    */
   static class Node {
@@ -82,62 +93,21 @@
     pathIdx = 0;
   }
 
-  /**
-   * Choose some files to RAID.
-   * @param conf Configuration to use.
-   * @param raidDestPrefix Prefix of the path to RAID to.
-   * @param modTimePeriod Time gap before RAIDing.
-   * @param limit Limit on the number of files to choose.
-   * @return list of files to RAID.
-   * @throws IOException
-   */
-  public List<FileStatus> selectFilesToRaid(
-      Configuration conf, int targetRepl, Path raidDestPrefix,
-      long modTimePeriod, int limit) throws IOException {
-    List<FileStatus> selected = new LinkedList<FileStatus>();
-    int numSelected = 0;
-
-    long now = System.currentTimeMillis();
-    while (numSelected < limit) {
+  public List<FileStatus> getFilteredFiles(FileFilter filter, int limit)
+      throws IOException {
+    List<FileStatus> filtered = new LinkedList<FileStatus>();
+    int num = 0;
+    while (num < limit) {
       FileStatus next = getNextFile();
       if (next == null) {
         break;
       }
-      // We have the next file, do we want to select it?
-      // If the source file has fewer than or equal to 2 blocks, then skip it.
-      long blockSize = next.getBlockSize();
-      if (2 * blockSize >= next.getLen()) {
-        continue;
-      }
-
-      boolean select = false;
-      try {
-        Object ppair = RaidNode.getParityFile(
-            raidDestPrefix, next.getPath(), conf);
-        // Is there is a valid parity file?
-        if (ppair != null) {
-          // Is the source at the target replication?
-          if (next.getReplication() != targetRepl) {
-            // Select the file so that its replication can be set.
-            select = true;
-          } else {
-            // Nothing to do, don't select the file.
-            select = false;
-          }
-        } else if (next.getModificationTime() + modTimePeriod < now) {
-          // If there isn't a valid parity file, check if the file is too new.
-          select = true;
-        }
-      } catch (java.io.FileNotFoundException e) {
-        select = true; // destination file does not exist
-      }
-      if (select) {
-        selected.add(next);
-        numSelected++;
+      if (filter.check(next)) {
+        num++;
+        filtered.add(next);
       }
     }
-
-    return selected;
+    return filtered;
   }
 
   /**
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java
new file mode 100644
index 0000000..2a660b4
--- /dev/null
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidFilter.java
@@ -0,0 +1,221 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.raid;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.raid.protocol.PolicyInfo;
+
+public class RaidFilter {
+  static class Statistics {
+    long numRaided = 0;
+    long numTooNew = 0;
+    long sizeTooNew = 0;
+    long numTooSmall = 0;
+    long sizeTooSmall = 0;
+
+    public void aggregate(Statistics other) {
+      this.numRaided += other.numRaided;
+      this.numTooNew += other.numTooNew;
+      this.sizeTooNew += other.sizeTooNew;
+      this.numTooSmall += other.numTooSmall;
+      this.sizeTooSmall += other.sizeTooSmall;
+    }
+
+    public String toString() {
+      return "numRaided = " + numRaided +
+             ", numTooNew = " + numTooNew +
+             ", sizeTooNew = " + sizeTooNew +
+             ", numTooSmall = " + numTooSmall +
+             ", sizeTooSmall = " + sizeTooSmall;
+    }
+  }
+
+  static class TimeBasedFilter extends Configured
+    implements DirectoryTraversal.FileFilter {
+    int targetRepl;
+    Path raidDestPrefix;
+    long modTimePeriod;
+    long startTime;
+    Statistics stats = new Statistics();
+    String currentSrcPath = null;
+    long[] modTimePeriods = new long[0];
+    String[] otherSrcPaths = new String[0];
+
+    TimeBasedFilter(Configuration conf, Path destPrefix, int targetRepl,
+      long startTime, long modTimePeriod) {
+      super(conf);
+      this.raidDestPrefix = destPrefix;
+      this.targetRepl = targetRepl;
+      this.startTime = startTime;
+      this.modTimePeriod = modTimePeriod;
+    }
+
+    TimeBasedFilter(Configuration conf,
+      Path destPrefix, PolicyInfo info,
+      List<PolicyInfo> allPolicies, long startTime, Statistics stats) {
+      super(conf);
+      this.raidDestPrefix = destPrefix;
+      this.targetRepl = Integer.parseInt(info.getProperty("targetReplication"));
+      this.modTimePeriod = Long.parseLong(info.getProperty("modTimePeriod"));
+      this.startTime = startTime;
+      this.stats = stats;
+      this.currentSrcPath = info.getSrcPath().toUri().getPath();
+      initializeOtherPaths(allPolicies);
+    }
+
+    private void initializeOtherPaths(List<PolicyInfo> allPolicies) {
+      ArrayList<PolicyInfo> tmp = new ArrayList<PolicyInfo>(allPolicies);
+      // Remove all policies where srcPath <= currentSrcPath or
+      // matchingPrefixLength is < length(currentSrcPath)
+      // The policies remaining are the only ones that could better
+      // select a file chosen by the current policy.
+      for (Iterator<PolicyInfo> it = tmp.iterator(); it.hasNext(); ) {
+        String src = it.next().getSrcPath().toUri().getPath();
+        if (src.compareTo(currentSrcPath) <= 0) {
+          it.remove();
+          continue;
+        }
+        int matchLen = matchingPrefixLength(src, currentSrcPath);
+        if (matchLen < currentSrcPath.length()) {
+          it.remove();
+        }
+      }
+      // Sort in reverse lexicographic order.
+      Collections.sort(tmp, new Comparator() {
+        public int compare(Object o1, Object o2) {
+          return 0 -
+            ((PolicyInfo)o1).getSrcPath().toUri().getPath().compareTo(
+              ((PolicyInfo)o1).getSrcPath().toUri().getPath());
+        }
+      });
+      otherSrcPaths = new String[tmp.size()];
+      modTimePeriods = new long[otherSrcPaths.length];
+      for (int i = 0; i < otherSrcPaths.length; i++) {
+        otherSrcPaths[i] = tmp.get(i).getSrcPath().toUri().getPath();
+        modTimePeriods[i] = Long.parseLong(
+          tmp.get(i).getProperty("modTimePeriod"));
+      }
+    }
+
+    public boolean check(FileStatus f) throws IOException {
+      if (!canChooseForCurrentPolicy(f)) {
+        return false;
+      }
+
+      // If the source file has fewer than or equal to 2 blocks, then skip it.
+      long blockSize = f.getBlockSize();
+      if (2 * blockSize >= f.getLen()) {
+        stats.numTooSmall++;
+        stats.sizeTooSmall += f.getLen();
+        return false;
+      }
+
+      boolean select = false;
+      try {
+        Object ppair = RaidNode.getParityFile(
+            raidDestPrefix, f.getPath(), getConf());
+        // Is there is a valid parity file?
+        if (ppair != null) {
+          // Is the source at the target replication?
+          if (f.getReplication() != targetRepl) {
+            // Select the file so that its replication can be set.
+            select = true;
+          } else {
+            stats.numRaided++;
+            // Nothing to do, don't select the file.
+            select = false;
+          }
+        } else {
+          // No parity file.
+          if (f.getModificationTime() + modTimePeriod < startTime) {
+            // If the file is not too new, choose it for raiding.
+            select = true;
+          } else {
+            select = false;
+            stats.numTooNew++;
+            stats.sizeTooNew += f.getLen();
+          }
+        }
+      } catch (java.io.FileNotFoundException e) {
+        select = true; // destination file does not exist
+      } catch (java.io.IOException e) {
+        // If there is a problem with the har path, this will let us continue.
+        DirectoryTraversal.LOG.error(
+          "Error while selecting " + StringUtils.stringifyException(e));
+      }
+      return select;
+    }
+
+    /**
+     * Checks if a file can be chosen for the current policy.
+     */
+    boolean canChooseForCurrentPolicy(FileStatus stat) {
+      boolean choose = true;
+      if (otherSrcPaths.length > 0) {
+        String fileStr = stat.getPath().toUri().getPath();
+
+        // For a given string, find the best matching srcPath.
+        int matchWithCurrent = matchingPrefixLength(fileStr, currentSrcPath);
+        for (int i = 0; i < otherSrcPaths.length; i++) {
+          // If the file is too new, move to the next.
+          if (stat.getModificationTime() > startTime - modTimePeriods[i]) {
+            continue;
+          }
+          int matchLen = matchingPrefixLength(fileStr, otherSrcPaths[i]);
+          if (matchLen > 0 &&
+              fileStr.charAt(matchLen - 1) == Path.SEPARATOR_CHAR) {
+            matchLen--;
+          }
+          if (matchLen > matchWithCurrent) {
+            choose = false;
+            break;
+          }
+        }
+      }
+      return choose;
+    }
+
+    int matchingPrefixLength(final String s1, final String s2) {
+      int len = 0;
+      for (int j = 0; j < s1.length() && j < s2.length(); j++) {
+        if (s1.charAt(j) == s2.charAt(j)) {
+          len++;
+        } else {
+          break;
+        }
+      }
+      return len;
+    }
+  }
+}
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
index 5507246..ec2f123 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/RaidNode.java
@@ -22,6 +22,7 @@
 import java.io.FileNotFoundException;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.LinkedList;
 import java.util.Iterator;
@@ -334,9 +335,19 @@
    */
   class TriggerMonitor implements Runnable {
 
-    private Map<String, Long> scanTimes = new HashMap<String, Long>();
-    private Map<String, DirectoryTraversal> scanState =
-      new HashMap<String, DirectoryTraversal>();
+    class ScanState {
+      long fullScanStartTime;
+      DirectoryTraversal pendingTraversal;
+      RaidFilter.Statistics stats;
+      ScanState() {
+        fullScanStartTime = 0;
+        pendingTraversal = null;
+        stats = new RaidFilter.Statistics();
+      }
+    }
+
+    private Map<String, ScanState> scanStateMap =
+      new HashMap<String, ScanState>();
 
     /**
      */
@@ -357,9 +368,10 @@
      */
     private boolean shouldSelectFiles(PolicyInfo info) {
       String policyName = info.getName();
+      ScanState scanState = scanStateMap.get(policyName);
       int runningJobsCount = jobMonitor.runningJobsCount(policyName);
       // Is there a scan in progress for this policy?
-      if (scanState.containsKey(policyName)) {
+      if (scanState.pendingTraversal != null) {
         int maxJobsPerPolicy = configMgr.getMaxJobsPerPolicy();
 
         // If there is a scan in progress for this policy, we can have
@@ -373,12 +385,8 @@
         }
         // Check the time of the last full traversal before starting a fresh
         // traversal.
-        if (scanTimes.containsKey(policyName)) {
-          long lastScan = scanTimes.get(policyName);
-          return (now() > lastScan + configMgr.getPeriodicity());
-        } else {
-          return true;
-        }
+        long lastScan = scanState.fullScanStartTime;
+        return (now() > lastScan + configMgr.getPeriodicity());
       }
     }
 
@@ -388,33 +396,29 @@
     * traversal.
     * The number of paths returned is limited by raid.distraid.max.jobs.
     */
-    private List<FileStatus> selectFiles(PolicyInfo info) throws IOException {
+    private List<FileStatus> selectFiles(
+      PolicyInfo info, ArrayList<PolicyInfo> allPolicies) throws IOException {
       Path destPrefix = getDestinationPath(conf);
       String policyName = info.getName();
       Path srcPath = info.getSrcPath();
-      long modTimePeriod = 0;
-      String str = info.getProperty("modTimePeriod");
-      if (str != null) {
-         modTimePeriod = Long.parseLong(info.getProperty("modTimePeriod"));
-      }
-      short srcReplication = 0;
-      str = info.getProperty("srcReplication");
-      if (str != null) {
-        srcReplication = Short.parseShort(info.getProperty("srcReplication"));
-      }
+      long modTimePeriod = Long.parseLong(info.getProperty("modTimePeriod"));
 
       // Max number of files returned.
       int selectLimit = configMgr.getMaxFilesPerJob();
       int targetRepl = Integer.parseInt(info.getProperty("targetReplication"));
 
+      long selectStartTime = System.currentTimeMillis();
+
+      ScanState scanState = scanStateMap.get(policyName);
       // If we have a pending traversal, resume it.
-      if (scanState.containsKey(policyName)) {
-        DirectoryTraversal dt = scanState.get(policyName);
+      if (scanState.pendingTraversal != null) {
+        DirectoryTraversal dt = scanState.pendingTraversal;
         LOG.info("Resuming traversal for policy " + policyName);
-        List<FileStatus> returnSet = dt.selectFilesToRaid(
-            conf, targetRepl, destPrefix, modTimePeriod, selectLimit);
+        DirectoryTraversal.FileFilter filter =
+          filterForPolicy(selectStartTime, info, allPolicies, scanState.stats);
+        List<FileStatus> returnSet = dt.getFilteredFiles(filter, selectLimit);
         if (dt.doneTraversal()) {
-          scanState.remove(policyName);
+          scanState.pendingTraversal = null;
         }
         return returnSet;
       }
@@ -445,12 +449,13 @@
         }
 
         // Set the time for a new traversal.
-        scanTimes.put(policyName, now());
+        scanState.fullScanStartTime = now();
         DirectoryTraversal dt = new DirectoryTraversal(fs, selectedPaths);
-        returnSet = dt.selectFilesToRaid(
-            conf, targetRepl, destPrefix, modTimePeriod, selectLimit);
+        DirectoryTraversal.FileFilter filter =
+          filterForPolicy(selectStartTime, info, allPolicies, scanState.stats);
+        returnSet = dt.getFilteredFiles(filter, selectLimit);
         if (!dt.doneTraversal()) {
-          scanState.put(policyName, dt);
+          scanState.pendingTraversal = dt;
         }
       }
       return returnSet;
@@ -461,73 +466,86 @@
      * If the config file has changed, then reload config file and start afresh.
      */
     private void doProcess() throws IOException, InterruptedException {
-      PolicyList.CompareByPath lexi = new PolicyList.CompareByPath();
-
+      ArrayList<PolicyInfo> allPolicies = new ArrayList<PolicyInfo>();
+      for (PolicyList category : configMgr.getAllPolicies()) {
+        for (PolicyInfo info: category.getAll()) {
+          allPolicies.add(info);
+        }
+      }
       while (running) {
         Thread.sleep(SLEEP_TIME);
 
-        configMgr.reloadConfigsIfNecessary();
-
-        // activate all categories
-        Collection<PolicyList> all = configMgr.getAllPolicies();
-        
-        // sort all policies by reverse lexicographical order. This is needed
-        // to make the nearest policy take precedence.
-        PolicyList[] sorted = all.toArray(new PolicyList[all.size()]);
-        Arrays.sort(sorted, lexi);
-
-        for (PolicyList category : sorted) {
-          for (PolicyInfo info: category.getAll()) {
-
-            if (!shouldSelectFiles(info)) {
-              continue;
-            }
-
-            LOG.info("Triggering Policy Filter " + info.getName() +
-                     " " + info.getSrcPath());
-            List<FileStatus> filteredPaths = null;
-            try {
-              filteredPaths = selectFiles(info);
-            } catch (Exception e) {
-              LOG.info("Exception while invoking filter on policy " + info.getName() +
-                       " srcPath " + info.getSrcPath() + 
-                       " exception " + StringUtils.stringifyException(e));
-              continue;
-            }
-
-            if (filteredPaths == null || filteredPaths.size() == 0) {
-              LOG.info("No filtered paths for policy " + info.getName());
-               continue;
-            }
-
-            // Apply the action on accepted paths
-            LOG.info("Triggering Policy Action " + info.getName() +
-                     " " + info.getSrcPath());
-            try {
-              if (isRaidLocal){
-                doRaid(conf, info, filteredPaths);
-              }
-              else{
-                // We already checked that no job for this policy is running
-                // So we can start a new job.
-                DistRaid dr = new DistRaid(conf);
-                //add paths for distributed raiding
-                dr.addRaidPaths(info, filteredPaths);
-                boolean started = dr.startDistRaid();
-                if (started) {
-                  jobMonitor.monitorJob(info.getName(), dr);
-                }
-              }
-            } catch (Exception e) {
-              LOG.info("Exception while invoking action on policy " + info.getName() +
-                       " srcPath " + info.getSrcPath() + 
-                       " exception " + StringUtils.stringifyException(e));
-              continue;
+        boolean reloaded = configMgr.reloadConfigsIfNecessary();
+        if (reloaded) {
+          allPolicies.clear();
+          for (PolicyList category : configMgr.getAllPolicies()) {
+            for (PolicyInfo info: category.getAll()) {
+              allPolicies.add(info);
             }
           }
         }
+
+        for (PolicyInfo info: allPolicies) {
+          if (!scanStateMap.containsKey(info.getName())) {
+            scanStateMap.put(info.getName(), new ScanState());
+          }
+
+          if (!shouldSelectFiles(info)) {
+            continue;
+          }
+
+          LOG.info("Triggering Policy Filter " + info.getName() +
+                   " " + info.getSrcPath());
+          List<FileStatus> filteredPaths = null;
+          try {
+            filteredPaths = selectFiles(info, allPolicies);
+          } catch (Exception e) {
+            LOG.info("Exception while invoking filter on policy " + info.getName() +
+                     " srcPath " + info.getSrcPath() + 
+                     " exception " + StringUtils.stringifyException(e));
+            continue;
+          }
+
+          if (filteredPaths == null || filteredPaths.size() == 0) {
+            LOG.info("No filtered paths for policy " + info.getName());
+             continue;
+          }
+
+          // Apply the action on accepted paths
+          LOG.info("Triggering Policy Action " + info.getName() +
+                   " " + info.getSrcPath());
+          try {
+            if (isRaidLocal){
+              doRaid(conf, info, filteredPaths);
+            }
+            else{
+              // We already checked that no job for this policy is running
+              // So we can start a new job.
+              DistRaid dr = new DistRaid(conf);
+              //add paths for distributed raiding
+              dr.addRaidPaths(info, filteredPaths);
+              boolean started = dr.startDistRaid();
+              if (started) {
+                jobMonitor.monitorJob(info.getName(), dr);
+              }
+            }
+          } catch (Exception e) {
+            LOG.info("Exception while invoking action on policy " + info.getName() +
+                     " srcPath " + info.getSrcPath() + 
+                     " exception " + StringUtils.stringifyException(e));
+            continue;
+          }
+        }
       }
     }
+
+    DirectoryTraversal.FileFilter filterForPolicy(
+      long startTime, PolicyInfo info, List<PolicyInfo> allPolicies,
+      RaidFilter.Statistics stats)
+      throws IOException {
+      return new RaidFilter.TimeBasedFilter(conf, getDestinationPath(conf),
+        info, allPolicies, startTime, stats);
+    }
   }
 
   static private Path getOriginalParityFile(Path destPathPrefix, Path srcPath) {
@@ -618,9 +636,9 @@
   }
   
   private ParityFilePair getParityFile(Path destPathPrefix, Path srcPath) throws IOException {
-	  
-	  return getParityFile(destPathPrefix, srcPath, conf);
-	  
+    
+    return getParityFile(destPathPrefix, srcPath, conf);
+    
   }
   
 
@@ -825,8 +843,6 @@
      * destination directories.
      */
     private void doPurge() throws IOException, InterruptedException {
-      PolicyList.CompareByPath lexi = new PolicyList.CompareByPath();
-
       long prevExec = 0;
       while (running) {
 
@@ -838,16 +854,8 @@
 
         LOG.info("Started purge scan");
         prevExec = now();
-        
-        // fetch all categories
-        Collection<PolicyList> all = configMgr.getAllPolicies();
-        
-        // sort all policies by reverse lexicographical order. This is 
-        // needed to make the nearest policy take precedence.
-        PolicyList[] sorted = all.toArray(new PolicyList[all.size()]);
-        Arrays.sort(sorted, lexi);
 
-        for (PolicyList category : sorted) {
+        for (PolicyList category : configMgr.getAllPolicies()) {
           for (PolicyInfo info: category.getAll()) {
 
             try {
@@ -857,7 +865,7 @@
 
               //get srcPaths
               Path[] srcPaths = info.getSrcPathExpanded();
-              
+
               if (srcPaths != null) {
                 for (Path srcPath: srcPaths) {
                   // expand destination prefix
@@ -1017,9 +1025,6 @@
   }
 
   private void doHar() throws IOException, InterruptedException {
-    
-    PolicyList.CompareByPath lexi = new PolicyList.CompareByPath();
-
     long prevExec = 0;
     while (running) {
 
@@ -1031,16 +1036,8 @@
 
       LOG.info("Started archive scan");
       prevExec = now();
-      
-      // fetch all categories
-      Collection<PolicyList> all = configMgr.getAllPolicies();
-            
-      // sort all policies by reverse lexicographical order. This is 
-      // needed to make the nearest policy take precedence.
-      PolicyList[] sorted = all.toArray(new PolicyList[all.size()]);
-      Arrays.sort(sorted, lexi);
 
-      for (PolicyList category : sorted) {
+      for (PolicyList category : configMgr.getAllPolicies()) {
         for (PolicyInfo info: category.getAll()) {
           String str = info.getProperty("time_before_har");
           String tmpHarPath = info.getProperty("har_tmp_dir");
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java
index c58eb8c..90e1ad4 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyInfo.java
@@ -137,7 +137,7 @@
   /**
    * Get the srcPath
    */
-  public Path getSrcPath() throws IOException {
+  public Path getSrcPath() {
     return srcPath;
   }
 
diff --git a/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyList.java b/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyList.java
index 187ed25..eb3dc6e 100644
--- a/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyList.java
+++ b/src/contrib/raid/src/java/org/apache/hadoop/raid/protocol/PolicyList.java
@@ -24,7 +24,6 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.LinkedList;
-import java.util.Comparator;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -78,18 +77,6 @@
     return category;
   }
 
-  /**
-   * Sort Categries based on their srcPath. reverse lexicographical order.
-   */
-  public static class CompareByPath implements Comparator<PolicyList> {
-    public CompareByPath() throws IOException {
-    }
-    public int compare(PolicyList l1, PolicyList l2) {
-      return 0 - l1.getSrcPath().compareTo(l2.getSrcPath());
-    }
-  }
-  
-  
   //////////////////////////////////////////////////
   // Writable
   //////////////////////////////////////////////////
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java
index 8a2805a..d7288ac 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestBlockFixer.java
@@ -239,18 +239,16 @@
       corruptBlock(locs.get(0).getBlock().getBlockName());
       reportCorruptBlocks(dfs, file1, new int[]{0}, blockSize);
 
-      // This should fail.
-      boolean caughtChecksumException = false;
       try {
         Thread.sleep(5*1000);
       } catch (InterruptedException ignore) {
       }
       try {
         TestRaidDfs.validateFile(dfs, file1, file1Len, crc1);
+        fail("Expected exception not thrown");
       } catch (org.apache.hadoop.fs.ChecksumException ce) {
-        caughtChecksumException = true;
+      } catch (org.apache.hadoop.hdfs.BlockMissingException bme) {
       }
-      assertTrue(caughtChecksumException);
     } catch (Exception e) {
       LOG.info("Test testGeneratedBlock Exception " + e + StringUtils.stringifyException(e));
       throw e;
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java
index 119cae6..9f7f710 100644
--- a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestDirectoryTraversal.java
@@ -32,6 +32,7 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.raid.protocol.PolicyInfo;
 
 public class TestDirectoryTraversal extends TestCase {
   final static Log LOG = LogFactory.getLog(
@@ -104,14 +105,16 @@
       int limit = 2;
       short targetRepl = 1;
       Path raid = new Path("/raid");
-      List<FileStatus> selected = dt.selectFilesToRaid(conf, targetRepl, raid,
-                                                        0, limit);
+      DirectoryTraversal.FileFilter filter =
+        new RaidFilter.TimeBasedFilter(conf,
+          RaidNode.getDestinationPath(conf), 1, System.currentTimeMillis(), 0);
+      List<FileStatus> selected = dt.getFilteredFiles(filter, limit);
       for (FileStatus f: selected) {
         LOG.info(f.getPath());
       }
       assertEquals(limit, selected.size());
 
-      selected = dt.selectFilesToRaid(conf, targetRepl, raid, 0, limit);
+      selected = dt.getFilteredFiles(filter, limit);
       for (FileStatus f: selected) {
         LOG.info(f.getPath());
       }
diff --git a/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java
new file mode 100644
index 0000000..9b1cdd9
--- /dev/null
+++ b/src/contrib/raid/src/test/org/apache/hadoop/raid/TestRaidFilter.java
@@ -0,0 +1,117 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.raid;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import junit.framework.TestCase;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.raid.protocol.PolicyInfo;
+
+public class TestRaidFilter extends TestCase {
+  final static String TEST_DIR = new File(System.getProperty("test.build.data",
+      "build/contrib/raid/test/data")).getAbsolutePath();
+  final static Log LOG =
+    LogFactory.getLog("org.apache.hadoop.raid.TestRaidFilter");
+
+  Configuration conf;
+  MiniDFSCluster dfs = null;
+  FileSystem fs = null;
+
+  private void mySetup() throws Exception {
+    new File(TEST_DIR).mkdirs(); // Make sure data directory exists
+    conf = new Configuration();
+    dfs = new MiniDFSCluster(conf, 2, true, null);
+    dfs.waitActive();
+    fs = dfs.getFileSystem();
+    String namenode = fs.getUri().toString();
+    FileSystem.setDefaultUri(conf, namenode);
+  }
+
+  private void myTearDown() throws Exception {
+    if (dfs != null) { dfs.shutdown(); }
+  }
+
+  public void testLayeredPolicies() throws Exception {
+    mySetup();
+    Path src1 = new Path("/user/foo");
+    Path src2 = new Path("/user/foo/bar");
+
+    PolicyInfo info1 = new PolicyInfo("p1", conf);
+    info1.setSrcPath(src1.toString());
+    info1.setDescription("test policy");
+    info1.setProperty("targetReplication", "1");
+    info1.setProperty("metaReplication", "1");
+    info1.setProperty("modTimePeriod", "0");
+
+    PolicyInfo info2 = new PolicyInfo("p2", conf);
+    info2.setSrcPath(src2.toString());
+    info2.setDescription("test policy");
+    info2.setProperty("targetReplication", "1");
+    info2.setProperty("metaReplication", "1");
+    info2.setProperty("modTimePeriod", "0");
+
+    ArrayList<PolicyInfo> all = new ArrayList<PolicyInfo>();
+    all.add(info1);
+    all.add(info2);
+
+    try {
+      long blockSize = 1024;
+      byte[] bytes = new byte[(int)blockSize];
+      Path f1 = new Path(src1, "f1");
+      Path f2 = new Path(src2, "f2");
+      FSDataOutputStream stm1 = fs.create(f1, false, 4096, (short)1, blockSize);
+      FSDataOutputStream stm2 = fs.create(f2, false, 4096, (short)1, blockSize);
+      FSDataOutputStream[]  stms = new FSDataOutputStream[]{stm1, stm2};
+      for (FSDataOutputStream stm: stms) {
+        stm.write(bytes);
+        stm.write(bytes);
+        stm.write(bytes);
+        stm.close();
+      }
+
+      Thread.sleep(1000);
+
+      FileStatus stat1 = fs.getFileStatus(f1);
+      FileStatus stat2 = fs.getFileStatus(f2);
+
+      RaidFilter.Statistics stats = new RaidFilter.Statistics();
+      RaidFilter.TimeBasedFilter filter = new RaidFilter.TimeBasedFilter(
+        conf, RaidNode.getDestinationPath(conf), info1, all,
+        System.currentTimeMillis(), stats);
+      System.out.println("Stats " + stats);
+
+      assertTrue(filter.check(stat1));
+      assertFalse(filter.check(stat2));
+
+    } finally {
+      myTearDown();
+    }
+  }
+}