From a36362271f837dfa067549ae1cc7fda627951761 Mon Sep 17 00:00:00 2001
From: Matthieu Martin <ma.tt.b.ma.rt.in+parquet@gmail.com>
Date: Thu, 19 Jun 2014 10:34:15 -0700
Subject: [PATCH 1/4] Use LRU caching for footers in ParquetInputFormat.

---
 parquet-hadoop/pom.xml                        |   6 +
 .../main/java/parquet/hadoop/LruCache.java    |  89 +++++++++++
 .../parquet/hadoop/ParquetInputFormat.java    | 104 ++++++++++++-
 .../java/parquet/hadoop/TestInputFormat.java  |  60 +++++++-
 .../java/parquet/hadoop/TestLruCache.java     | 144 ++++++++++++++++++
 5 files changed, 393 insertions(+), 10 deletions(-)
 create mode 100644 parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
 create mode 100644 parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java

diff --git a/parquet-hadoop/pom.xml b/parquet-hadoop/pom.xml
index ec7b94d2de..c1b4f0770a 100644
--- a/parquet-hadoop/pom.xml
+++ b/parquet-hadoop/pom.xml
@@ -68,6 +68,12 @@
       <type>jar</type>
       <scope>compile</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <version>1.9.5</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
new file mode 100644
index 0000000000..92570ff3b0
--- /dev/null
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
@@ -0,0 +1,89 @@
+package parquet.hadoop;
+
+import parquet.Log;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+final class LruCache<K, V extends LruCache.Entry<V>> {
+  private static final Log LOG = Log.getLog(LruCache.class);
+
+  private static final float DEFAULT_LOAD_FACTOR = 0.75f;
+
+  private final LinkedHashMap<K, V> cacheMap;
+
+  public LruCache(int maxSize) {
+    this(maxSize, DEFAULT_LOAD_FACTOR, true);
+  }
+
+  public LruCache(final int maxSize, float loadFactor, boolean accessOrder) {
+    cacheMap =
+            new LinkedHashMap<K, V>(Math.round(maxSize / loadFactor), loadFactor, accessOrder) {
+              @Override
+              public boolean removeEldestEntry(Map.Entry<K,V> eldest) {
+                boolean result = size() > maxSize;
+                if (result) {
+                  if (Log.DEBUG) LOG.debug("Removing eldest entry in cache: " + eldest.getKey());
+                }
+                return result;
+              }
+            };
+  }
+
+  public V remove(K key) {
+    V oldEntry = cacheMap.remove(key);
+    if (oldEntry != null) {
+      if (Log.DEBUG) LOG.debug("Removed cache entry for '" + key + "'");
+    }
+    return oldEntry;
+  }
+
+  public void put(K key, V newEntry) {
+    if (newEntry == null || !newEntry.isCurrent()) {
+      if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because it is " + (newEntry == null ? "null" : "not current"));
+      return;
+    }
+
+    V oldEntry = cacheMap.get(key);
+    if (oldEntry != null && oldEntry.isNewerThan(newEntry)) {
+      if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because existing cache entry is newer");
+      return;
+    }
+
+    // no existing entry or new entry is newer than old entry
+    oldEntry = cacheMap.put(key, newEntry);
+    if (Log.DEBUG) {
+      if (oldEntry == null) {
+        LOG.debug("Added new cache entry for '" + key + "'");
+      } else {
+        LOG.debug("Overwrote existing cache entry for '" + key + "'");
+      }
+    }
+  }
+
+  public void clear() {
+    cacheMap.clear();
+  }
+
+  public V getCurrentEntry(K key) {
+    V value = cacheMap.get(key);
+    if (Log.DEBUG) LOG.debug("Entry for '" + key + "' " + (value == null ? "not " : "") + "in cache");
+    if (value != null && !value.isCurrent()) {
+      // value is not current; remove it and return null
+      remove(key);
+      return null;
+    }
+
+    return value;
+  }
+
+  public int size() {
+    return cacheMap.size();
+  }
+
+  interface Entry<V> {
+    public boolean isCurrent();
+    public boolean isNewerThan(V otherEntry);
+  }
+
+}
diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
index 3abb38b5d7..3f85d734ce 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
@@ -15,12 +15,17 @@
  */
 package parquet.hadoop;
 
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
@@ -78,8 +83,11 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
    */
   public static final String UNBOUND_RECORD_FILTER = "parquet.read.filter";
 
+  private static final int MIN_FOOTER_CACHE_SIZE = 100;
+
+  private LruCache<Path, FootersCacheEntry> footersCache;
+
   private Class<?> readSupportClass;
-  private List<Footer> footers;
 
   public static void setReadSupportClass(Job job,  Class<?> readSupportClass) {
     ContextUtil.getConfiguration(job).set(READ_SUPPORT_CLASS, readSupportClass.getName());
@@ -399,9 +407,7 @@ protected List<FileStatus> listStatus(JobContext jobContext) throws IOException
   private static List<FileStatus> getAllFileRecursively(
       List<FileStatus> files, Configuration conf) throws IOException {
     List<FileStatus> result = new ArrayList<FileStatus>();
-    int len = files.size();
-    for (int i = 0; i < len; ++i) {
-      FileStatus file = files.get(i);
+    for (FileStatus file : files) {
       if (file.isDir()) {
         Path p = file.getPath();
         FileSystem fs = p.getFileSystem(conf);
@@ -439,10 +445,52 @@ public boolean accept(Path p){
    * @throws IOException
    */
   public List<Footer> getFooters(JobContext jobContext) throws IOException {
-    if (footers == null) {
-      footers = getFooters(ContextUtil.getConfiguration(jobContext), listStatus(jobContext));
+    List<FileStatus> statuses = listStatus(jobContext);
+    if (statuses.isEmpty()) {
+      return Collections.emptyList();
+    }
+
+    Configuration config = ContextUtil.getConfiguration(jobContext);
+    List<Footer> footers = new ArrayList<Footer>(statuses.size());
+    Set<FileStatus> missingStatuses = new HashSet<FileStatus>();
+
+    if (footersCache != null) {
+      for (FileStatus status : statuses) {
+        FootersCacheEntry cacheEntry = footersCache.getCurrentEntry(status.getPath());
+        if (Log.DEBUG) LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "") + " found for '" + status.getPath() + "'");
+        if (cacheEntry != null) {
+          footers.add(cacheEntry.getFooter());
+        } else {
+          missingStatuses.add(status);
+        }
+      }
+    } else {
+      // initialize the cache to store all of the current statuses or a default minimum size. The sizing of this LRU
+      // cache was chosen to mimic prior behavior (i.e. so that performance would be at least as good as it was before)
+      footersCache = new LruCache<Path, FootersCacheEntry>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
+      missingStatuses.addAll(statuses);
     }
+    if (Log.DEBUG) LOG.debug("found " + footers.size() + " footers in cache and adding up to " +
+            missingStatuses.size() + " missing footers to the cache");
 
+
+    if (missingStatuses.isEmpty()) {
+      return footers;
+    }
+
+    List<Footer> newFooters = getFooters(config, new ArrayList<FileStatus>(missingStatuses));
+    Map<Path, FileStatus> missingStatusesMap = new HashMap<Path, FileStatus>(missingStatuses.size());
+    for (FileStatus missingStatus : missingStatuses) {
+      missingStatusesMap.put(missingStatus.getPath(), missingStatus);
+    }
+    for (Footer newFooter : newFooters) {
+      // Use the original file status objects to make sure we store a conservative (older) modification time (i.e. in
+      // case the files and footers were modified and it's not clear which version of the footers we have)
+      FileStatus fileStatus = missingStatusesMap.get(newFooter.getFile());
+      footersCache.put(fileStatus.getPath(), new FootersCacheEntry(fileStatus, newFooter));
+    }
+
+    footers.addAll(newFooters);
     return footers;
   }
 
@@ -454,7 +502,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
    * @throws IOException
    */
   public List<Footer> getFooters(Configuration configuration, List<FileStatus> statuses) throws IOException {
-    LOG.debug("reading " + statuses.size() + " files");
+    if (Log.DEBUG) LOG.debug("reading " + statuses.size() + " files");
     return ParquetFileReader.readAllFootersInParallelUsingSummaryFiles(configuration, statuses);
   }
 
@@ -467,4 +515,46 @@ public GlobalMetaData getGlobalMetaData(JobContext jobContext) throws IOExceptio
     return ParquetFileWriter.getGlobalMetaData(getFooters(jobContext));
   }
 
+  static final class FootersCacheEntry implements LruCache.Entry<FootersCacheEntry> {
+    private final long modificationTime;
+    private final Footer footer;
+
+    public FootersCacheEntry(FileStatus status, Footer footer) {
+      this.modificationTime = status.getModificationTime();
+      this.footer = new Footer(footer.getFile(), footer.getParquetMetadata());
+    }
+
+    public boolean isCurrent() {
+      FileSystem fs;
+      FileStatus currentFile;
+      Path path = footer.getFile();
+      try {
+        fs = path.getFileSystem(new Configuration());
+        currentFile = fs.getFileStatus(path);
+      } catch (FileNotFoundException e) {
+        if (Log.DEBUG) LOG.debug("The '" + path + "' path was not found.");
+        return false;
+      } catch (IOException e) {
+        throw new RuntimeException("Exception while checking '" + path + "': " + e, e);
+      }
+      long currentModTime = currentFile.getModificationTime();
+      boolean isCurrent = modificationTime >= currentModTime;
+      if (Log.DEBUG && !isCurrent) LOG.debug("The cache entry for '" + currentFile.getPath() + "' is not current: " +
+              "cached modification time=" + modificationTime + ", current modification time: " + currentModTime);
+      return isCurrent;
+    }
+
+    public Footer getFooter() {
+      return footer;
+    }
+
+    public boolean isNewerThan(FootersCacheEntry entry) {
+      return entry == null || modificationTime > entry.modificationTime;
+    }
+
+    public Path getPath() {
+      return footer.getFile();
+    }
+  }
+
 }
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
index 989869c5a4..ad935e8cb0 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
@@ -15,24 +15,27 @@
  */
 package parquet.hadoop;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.junit.Before;
 import org.junit.Test;
 import parquet.column.Encoding;
 import parquet.column.statistics.BinaryStatistics;
-import parquet.hadoop.api.ReadSupport;
 import parquet.hadoop.metadata.BlockMetaData;
 import parquet.hadoop.metadata.ColumnChunkMetaData;
 import parquet.hadoop.metadata.ColumnPath;
 import parquet.hadoop.metadata.CompressionCodecName;
 import parquet.hadoop.metadata.FileMetaData;
+import parquet.hadoop.metadata.ParquetMetadata;
 import parquet.io.ParquetDecodingException;
 import parquet.schema.MessageType;
 import parquet.schema.MessageTypeParser;
 import parquet.schema.PrimitiveType.PrimitiveTypeName;
 
+import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -40,8 +43,8 @@
 import java.util.List;
 import java.util.Arrays;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
 
 public class TestInputFormat {
 
@@ -243,6 +246,57 @@ public void testMultipleRowGroupsInABlockToAlignHDFSBlock() throws Exception {
     shouldSplitLengthBe(splits, 20, 20, 10, 20, 20, 10);
   }
 
+  @Test
+  public void testFooterCacheEntryIsCurrent() throws IOException, InterruptedException {
+    File tempFile = getTempFile();
+    FileSystem fs = FileSystem.getLocal(new Configuration());
+    ParquetInputFormat.FootersCacheEntry cacheEntry = getDummyCacheEntry(tempFile, fs);
+
+    assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
+    assertFalse(cacheEntry.isCurrent());
+  }
+
+  @Test
+  public void testFooterCacheEntryDeleted() throws IOException {
+    File tempFile = getTempFile();
+    FileSystem fs = FileSystem.getLocal(new Configuration());
+    ParquetInputFormat.FootersCacheEntry cacheEntry = getDummyCacheEntry(tempFile, fs);
+
+    assertTrue(tempFile.delete());
+    assertFalse(cacheEntry.isCurrent());
+  }
+
+  @Test
+  public void testFooterCacheEntryIsNewer() throws IOException {
+    File tempFile = getTempFile();
+    FileSystem fs = FileSystem.getLocal(new Configuration());
+    ParquetInputFormat.FootersCacheEntry cacheEntry = getDummyCacheEntry(tempFile, fs);
+
+    assertTrue(cacheEntry.isNewerThan(null));
+    assertFalse(cacheEntry.isNewerThan(cacheEntry));
+
+    assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
+    ParquetInputFormat.FootersCacheEntry newerCacheEntry = getDummyCacheEntry(tempFile, fs);
+
+    assertTrue(newerCacheEntry.isNewerThan(cacheEntry));
+    assertFalse(cacheEntry.isNewerThan(newerCacheEntry));
+  }
+
+  private File getTempFile() throws IOException {
+    File tempFile = File.createTempFile("footer_", ".txt");
+    tempFile.deleteOnExit();
+    return tempFile;
+  }
+
+  private ParquetInputFormat.FootersCacheEntry getDummyCacheEntry(File file, FileSystem fs) throws IOException {
+    Path path = new Path(file.getPath());
+    FileStatus status = fs.getFileStatus(path);
+    ParquetMetadata mockMetadata = mock(ParquetMetadata.class);
+    ParquetInputFormat.FootersCacheEntry cacheEntry = new ParquetInputFormat.FootersCacheEntry(status, new Footer(path, mockMetadata));
+    assertTrue(cacheEntry.isCurrent());
+    return cacheEntry;
+  }
+
   private List<ParquetInputSplit> generateSplitByMinMaxSize(long min, long max) throws IOException {
     return ParquetInputFormat.generateSplits(
             blocks, hdfsBlocks, fileStatus, fileMetaData, schema.toString(), new HashMap<String, String>() {{
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
new file mode 100644
index 0000000000..baf023c1a1
--- /dev/null
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
@@ -0,0 +1,144 @@
+package parquet.hadoop;
+
+import org.junit.Test;
+
+import static org.junit.Assert.*;
+
+public class TestLruCache {
+  private static final String DEFAULT_KEY = "test";
+
+  private static final class SimpleEntry implements LruCache.Entry<SimpleEntry> {
+    private boolean current;
+    private boolean newerThan;
+
+    public SimpleEntry(boolean current, boolean newerThan) {
+      this.current = current;
+      this.newerThan = newerThan;
+    }
+
+    @Override
+    public boolean isCurrent() {
+      return current;
+    }
+
+    public void setCurrent(boolean current) {
+      this.current = current;
+    }
+
+    @Override
+    public boolean isNewerThan(SimpleEntry otherEntry) {
+      return newerThan;
+    }
+
+  }
+
+  @Test
+  public void testMaxSize() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+
+    String oldKey = DEFAULT_KEY;
+    String newKey = oldKey + "_new";
+
+    SimpleEntry oldEntry = new SimpleEntry(true, true);
+    cache.put(oldKey, oldEntry);
+    assertEquals(oldEntry, cache.getCurrentEntry(oldKey));
+    assertEquals(1, cache.size());
+
+    SimpleEntry newEntry = new SimpleEntry(true, true);
+    cache.put(newKey, newEntry);
+    assertNull(cache.getCurrentEntry(oldKey));
+    assertEquals(newEntry, cache.getCurrentEntry(newKey));
+    assertEquals(1, cache.size());
+  }
+
+  @Test
+  public void testOlderEntryIsIgnored() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+
+    SimpleEntry currentEntry = new SimpleEntry(true, true);
+    SimpleEntry notAsCurrentEntry = new SimpleEntry(true, false);
+    cache.put(DEFAULT_KEY, currentEntry);
+    cache.put(DEFAULT_KEY, notAsCurrentEntry);
+    assertEquals(
+            "The existing entry in the cache was overwritten",
+            currentEntry,
+            cache.getCurrentEntry(DEFAULT_KEY)
+    );
+  }
+
+  @Test
+  public void testOutdatedEntryIsIgnored() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+
+    SimpleEntry outdatedEntry = new SimpleEntry(false, true);
+    cache.put(DEFAULT_KEY, outdatedEntry);
+    assertEquals(0, cache.size());
+    assertNull(cache.getCurrentEntry(DEFAULT_KEY));
+  }
+
+  @Test
+  public void testCurrentEntryOverwritesExisting() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+
+    SimpleEntry currentEntry = new SimpleEntry(true, true);
+    SimpleEntry notAsCurrentEntry = new SimpleEntry(true, false);
+    cache.put(DEFAULT_KEY, notAsCurrentEntry);
+    assertEquals(1, cache.size());
+    cache.put(DEFAULT_KEY, currentEntry);
+    assertEquals(1, cache.size());
+    assertEquals(
+            "The existing entry in the cache was NOT overwritten",
+            currentEntry,
+            cache.getCurrentEntry(DEFAULT_KEY)
+    );
+  }
+
+  @Test
+  public void testGetOutdatedEntryReturnsNull() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+
+    SimpleEntry entry = new SimpleEntry(true, true);
+    cache.put(DEFAULT_KEY, entry);
+    assertEquals(1, cache.size());
+    assertEquals(entry, cache.getCurrentEntry(DEFAULT_KEY));
+
+    entry.setCurrent(false);
+    assertNull("The entry should not be current anymore", cache.getCurrentEntry(DEFAULT_KEY));
+    assertEquals(0, cache.size());
+  }
+
+  @Test
+  public void testRemove() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+
+    SimpleEntry entry = new SimpleEntry(true, true);
+    cache.put(DEFAULT_KEY, entry);
+    assertEquals(1, cache.size());
+    assertEquals(entry, cache.getCurrentEntry(DEFAULT_KEY));
+
+    // remove the only entry
+    assertEquals(entry, cache.remove(DEFAULT_KEY));
+    assertNull(cache.getCurrentEntry(DEFAULT_KEY));
+    assertEquals(0, cache.size());
+  }
+
+  @Test
+  public void testClear() {
+    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(2);
+
+    String key1 = DEFAULT_KEY + 1;
+    String key2 = DEFAULT_KEY + 2;
+    SimpleEntry entry = new SimpleEntry(true, true);
+    cache.put(key1, entry);
+    cache.put(key2, entry);
+    assertEquals(entry, cache.getCurrentEntry(key1));
+    assertEquals(entry, cache.getCurrentEntry(key2));
+    assertEquals(2, cache.size());
+
+    cache.clear();
+    assertNull(cache.getCurrentEntry(key1));
+    assertNull(cache.getCurrentEntry(key2));
+    assertEquals(0, cache.size());
+  }
+
+}

From d94644543671ab9805322d23d315ff54a54e55b0 Mon Sep 17 00:00:00 2001
From: Matthieu Martin <ma.tt.b.ma.rt.in+parquet@gmail.com>
Date: Sun, 22 Jun 2014 11:16:49 -0700
Subject: [PATCH 2/4] Add javadocs to parquet.hadoop.LruCache.  Rename cache
 "entries" as cache "values" to avoid confusion with java.util.Map.Entry
 (which contains key value pairs whereas our old "entries" really only refer
 to the values).

---
 .../main/java/parquet/hadoop/LruCache.java    |  98 ++++++++++++---
 .../parquet/hadoop/ParquetInputFormat.java    |  18 +--
 .../java/parquet/hadoop/TestInputFormat.java  |  34 +++---
 .../java/parquet/hadoop/TestLruCache.java     | 112 +++++++++---------
 4 files changed, 164 insertions(+), 98 deletions(-)

diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
index 92570ff3b0..a8edc3d006 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
@@ -5,17 +5,37 @@
 import java.util.LinkedHashMap;
 import java.util.Map;
 
-final class LruCache<K, V extends LruCache.Entry<V>> {
+/**
+ * A basic implementation of an LRU cache.  Besides evicting the least recently used entries (either based on insertion
+ * or access order), this class also checks for "stale" entries as entries are inserted or retrieved (note "staleness"
+ * is defined by the entries themselves (see {@link parquet.hadoop.LruCache.Value}.
+ *
+ * @param <K> The key type. Acts as the key in a {@link java.util.LinkedHashMap}
+ * @param <V> The value type.  Must extend {@link parquet.hadoop.LruCache.Value} so that the "staleness" of the value
+ *           can be easily determined.
+ */
+final class LruCache<K, V extends LruCache.Value<V>> {
   private static final Log LOG = Log.getLog(LruCache.class);
 
   private static final float DEFAULT_LOAD_FACTOR = 0.75f;
 
   private final LinkedHashMap<K, V> cacheMap;
 
+  /**
+   * Constructs an access-order based LRU cache with {@code maxSize} entries.
+   * @param maxSize The maximum number of entries to store in the cache.
+   */
   public LruCache(int maxSize) {
     this(maxSize, DEFAULT_LOAD_FACTOR, true);
   }
 
+  /**
+   * Constructs an LRU cache.
+   *
+   * @param maxSize The maximum number of entries to store in the cache.
+   * @param loadFactor Used to determine the initial capacity.
+   * @param accessOrder the ordering mode - {@code true} for access-order, {@code false} for insertion-order
+   */
   public LruCache(final int maxSize, float loadFactor, boolean accessOrder) {
     cacheMap =
             new LinkedHashMap<K, V>(Math.round(maxSize / loadFactor), loadFactor, accessOrder) {
@@ -30,30 +50,43 @@ public boolean removeEldestEntry(Map.Entry<K,V> eldest) {
             };
   }
 
+  /**
+   * Removes the mapping for the specified key from this cache if present.
+   * @param key key whose mapping is to be removed from the cache
+   * @return the previous value associated with key, or null if there was no mapping for key.
+   */
   public V remove(K key) {
-    V oldEntry = cacheMap.remove(key);
-    if (oldEntry != null) {
+    V oldValue = cacheMap.remove(key);
+    if (oldValue != null) {
       if (Log.DEBUG) LOG.debug("Removed cache entry for '" + key + "'");
     }
-    return oldEntry;
+    return oldValue;
   }
 
-  public void put(K key, V newEntry) {
-    if (newEntry == null || !newEntry.isCurrent()) {
-      if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because it is " + (newEntry == null ? "null" : "not current"));
+  /**
+   * Associates the specified value with the specified key in this cache. The value is only inserted if it is not null
+   * and it is considered current. If the cache previously contained a mapping for the key, the old value is replaced
+   * only if the new value is "newer" than the old one.
+   * @param key key with which the specified value is to be associated
+   * @param newValue value to be associated with the specified key
+   */
+  public void put(K key, V newValue) {
+    if (newValue == null || !newValue.isCurrent()) {
+      if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because it is " +
+              (newValue == null ? "null" : "not current"));
       return;
     }
 
-    V oldEntry = cacheMap.get(key);
-    if (oldEntry != null && oldEntry.isNewerThan(newEntry)) {
+    V oldValue = cacheMap.get(key);
+    if (oldValue != null && oldValue.isNewerThan(newValue)) {
       if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because existing cache entry is newer");
       return;
     }
 
-    // no existing entry or new entry is newer than old entry
-    oldEntry = cacheMap.put(key, newEntry);
+    // no existing value or new value is newer than old value
+    oldValue = cacheMap.put(key, newValue);
     if (Log.DEBUG) {
-      if (oldEntry == null) {
+      if (oldValue == null) {
         LOG.debug("Added new cache entry for '" + key + "'");
       } else {
         LOG.debug("Overwrote existing cache entry for '" + key + "'");
@@ -61,13 +94,23 @@ public void put(K key, V newEntry) {
     }
   }
 
+  /**
+   * Removes all of the mappings from this cache. The cache will be empty after this call returns.
+   */
   public void clear() {
     cacheMap.clear();
   }
 
-  public V getCurrentEntry(K key) {
+  /**
+   * Returns the value to which the specified key is mapped, or null if 1) the value is not current or 2) this cache
+   * contains no mapping for the key.
+   * @param key the key whose associated value is to be returned
+   * @return the value to which the specified key is mapped, or null if 1) the value is not current or 2) this cache
+   * contains no mapping for the key
+   */
+  public V getCurrentValue(K key) {
     V value = cacheMap.get(key);
-    if (Log.DEBUG) LOG.debug("Entry for '" + key + "' " + (value == null ? "not " : "") + "in cache");
+    if (Log.DEBUG) LOG.debug("Value for '" + key + "' " + (value == null ? "not " : "") + "in cache");
     if (value != null && !value.isCurrent()) {
       // value is not current; remove it and return null
       remove(key);
@@ -77,13 +120,36 @@ public V getCurrentEntry(K key) {
     return value;
   }
 
+  /**
+   * Returns the number of key-value mappings in this cache.
+   * @return the number of key-value mappings in this cache.
+   */
   public int size() {
     return cacheMap.size();
   }
 
-  interface Entry<V> {
+  /**
+   * {@link parquet.hadoop.LruCache} expects all values to follow this interface so the cache can determine
+   * 1) whether values are current (e.g. the referenced data has not been modified/updated in such a way that the value
+   * is no longer useful) and 2) whether a value is strictly "newer" than another value.
+   *
+   * @param <V>
+   */
+  interface Value<V> {
+    /**
+     * Is the value still current (e.g. has the referenced data been modified/updated in such a way that the value
+     * is no longer useful)
+     * @return {@code true} the value is still current, {@code false} the value is no longer useful
+     */
     public boolean isCurrent();
-    public boolean isNewerThan(V otherEntry);
+
+    /**
+     * Compares this value with the specified value to check for relative age.
+     * @param otherValue the value to be compared.
+     * @return {@code true} the value is strictly newer than the other value, {@code false} the value is older or just
+     * as new as the other value.
+     */
+    public boolean isNewerThan(V otherValue);
   }
 
 }
diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
index 3f85d734ce..73943440ad 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
@@ -85,7 +85,7 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
 
   private static final int MIN_FOOTER_CACHE_SIZE = 100;
 
-  private LruCache<Path, FootersCacheEntry> footersCache;
+  private LruCache<Path, FootersCacheValue> footersCache;
 
   private Class<?> readSupportClass;
 
@@ -456,7 +456,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
 
     if (footersCache != null) {
       for (FileStatus status : statuses) {
-        FootersCacheEntry cacheEntry = footersCache.getCurrentEntry(status.getPath());
+        FootersCacheValue cacheEntry = footersCache.getCurrentValue(status.getPath());
         if (Log.DEBUG) LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "") + " found for '" + status.getPath() + "'");
         if (cacheEntry != null) {
           footers.add(cacheEntry.getFooter());
@@ -467,7 +467,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
     } else {
       // initialize the cache to store all of the current statuses or a default minimum size. The sizing of this LRU
       // cache was chosen to mimic prior behavior (i.e. so that performance would be at least as good as it was before)
-      footersCache = new LruCache<Path, FootersCacheEntry>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
+      footersCache = new LruCache<Path, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
       missingStatuses.addAll(statuses);
     }
     if (Log.DEBUG) LOG.debug("found " + footers.size() + " footers in cache and adding up to " +
@@ -487,7 +487,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
       // Use the original file status objects to make sure we store a conservative (older) modification time (i.e. in
       // case the files and footers were modified and it's not clear which version of the footers we have)
       FileStatus fileStatus = missingStatusesMap.get(newFooter.getFile());
-      footersCache.put(fileStatus.getPath(), new FootersCacheEntry(fileStatus, newFooter));
+      footersCache.put(fileStatus.getPath(), new FootersCacheValue(fileStatus, newFooter));
     }
 
     footers.addAll(newFooters);
@@ -515,11 +515,11 @@ public GlobalMetaData getGlobalMetaData(JobContext jobContext) throws IOExceptio
     return ParquetFileWriter.getGlobalMetaData(getFooters(jobContext));
   }
 
-  static final class FootersCacheEntry implements LruCache.Entry<FootersCacheEntry> {
+  static final class FootersCacheValue implements LruCache.Value<FootersCacheValue> {
     private final long modificationTime;
     private final Footer footer;
 
-    public FootersCacheEntry(FileStatus status, Footer footer) {
+    public FootersCacheValue(FileStatus status, Footer footer) {
       this.modificationTime = status.getModificationTime();
       this.footer = new Footer(footer.getFile(), footer.getParquetMetadata());
     }
@@ -539,7 +539,7 @@ public boolean isCurrent() {
       }
       long currentModTime = currentFile.getModificationTime();
       boolean isCurrent = modificationTime >= currentModTime;
-      if (Log.DEBUG && !isCurrent) LOG.debug("The cache entry for '" + currentFile.getPath() + "' is not current: " +
+      if (Log.DEBUG && !isCurrent) LOG.debug("The cache value for '" + currentFile.getPath() + "' is not current: " +
               "cached modification time=" + modificationTime + ", current modification time: " + currentModTime);
       return isCurrent;
     }
@@ -548,8 +548,8 @@ public Footer getFooter() {
       return footer;
     }
 
-    public boolean isNewerThan(FootersCacheEntry entry) {
-      return entry == null || modificationTime > entry.modificationTime;
+    public boolean isNewerThan(FootersCacheValue otherValue) {
+      return otherValue == null || modificationTime > otherValue.modificationTime;
     }
 
     public Path getPath() {
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
index ad935e8cb0..67fa73a418 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
@@ -247,39 +247,39 @@ public void testMultipleRowGroupsInABlockToAlignHDFSBlock() throws Exception {
   }
 
   @Test
-  public void testFooterCacheEntryIsCurrent() throws IOException, InterruptedException {
+  public void testFooterCacheValueIsCurrent() throws IOException, InterruptedException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.FootersCacheEntry cacheEntry = getDummyCacheEntry(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
-    assertFalse(cacheEntry.isCurrent());
+    assertFalse(cacheValue.isCurrent());
   }
 
   @Test
-  public void testFooterCacheEntryDeleted() throws IOException {
+  public void testFooterCacheValueDeleted() throws IOException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.FootersCacheEntry cacheEntry = getDummyCacheEntry(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(tempFile.delete());
-    assertFalse(cacheEntry.isCurrent());
+    assertFalse(cacheValue.isCurrent());
   }
 
   @Test
-  public void testFooterCacheEntryIsNewer() throws IOException {
+  public void testFooterCacheValueIsNewer() throws IOException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.FootersCacheEntry cacheEntry = getDummyCacheEntry(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
-    assertTrue(cacheEntry.isNewerThan(null));
-    assertFalse(cacheEntry.isNewerThan(cacheEntry));
+    assertTrue(cacheValue.isNewerThan(null));
+    assertFalse(cacheValue.isNewerThan(cacheValue));
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
-    ParquetInputFormat.FootersCacheEntry newerCacheEntry = getDummyCacheEntry(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue newerCacheValue = getDummyCacheValue(tempFile, fs);
 
-    assertTrue(newerCacheEntry.isNewerThan(cacheEntry));
-    assertFalse(cacheEntry.isNewerThan(newerCacheEntry));
+    assertTrue(newerCacheValue.isNewerThan(cacheValue));
+    assertFalse(cacheValue.isNewerThan(newerCacheValue));
   }
 
   private File getTempFile() throws IOException {
@@ -288,13 +288,13 @@ private File getTempFile() throws IOException {
     return tempFile;
   }
 
-  private ParquetInputFormat.FootersCacheEntry getDummyCacheEntry(File file, FileSystem fs) throws IOException {
+  private ParquetInputFormat.FootersCacheValue getDummyCacheValue(File file, FileSystem fs) throws IOException {
     Path path = new Path(file.getPath());
     FileStatus status = fs.getFileStatus(path);
     ParquetMetadata mockMetadata = mock(ParquetMetadata.class);
-    ParquetInputFormat.FootersCacheEntry cacheEntry = new ParquetInputFormat.FootersCacheEntry(status, new Footer(path, mockMetadata));
-    assertTrue(cacheEntry.isCurrent());
-    return cacheEntry;
+    ParquetInputFormat.FootersCacheValue cacheValue = new ParquetInputFormat.FootersCacheValue(status, new Footer(path, mockMetadata));
+    assertTrue(cacheValue.isCurrent());
+    return cacheValue;
   }
 
   private List<ParquetInputSplit> generateSplitByMinMaxSize(long min, long max) throws IOException {
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
index baf023c1a1..7081d54f8f 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
@@ -7,11 +7,11 @@
 public class TestLruCache {
   private static final String DEFAULT_KEY = "test";
 
-  private static final class SimpleEntry implements LruCache.Entry<SimpleEntry> {
+  private static final class SimpleValue implements LruCache.Value<SimpleValue> {
     private boolean current;
     private boolean newerThan;
 
-    public SimpleEntry(boolean current, boolean newerThan) {
+    public SimpleValue(boolean current, boolean newerThan) {
       this.current = current;
       this.newerThan = newerThan;
     }
@@ -26,7 +26,7 @@ public void setCurrent(boolean current) {
     }
 
     @Override
-    public boolean isNewerThan(SimpleEntry otherEntry) {
+    public boolean isNewerThan(SimpleValue otherValue) {
       return newerThan;
     }
 
@@ -34,110 +34,110 @@ public boolean isNewerThan(SimpleEntry otherEntry) {
 
   @Test
   public void testMaxSize() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(1);
 
     String oldKey = DEFAULT_KEY;
     String newKey = oldKey + "_new";
 
-    SimpleEntry oldEntry = new SimpleEntry(true, true);
-    cache.put(oldKey, oldEntry);
-    assertEquals(oldEntry, cache.getCurrentEntry(oldKey));
+    SimpleValue oldValue = new SimpleValue(true, true);
+    cache.put(oldKey, oldValue);
+    assertEquals(oldValue, cache.getCurrentValue(oldKey));
     assertEquals(1, cache.size());
 
-    SimpleEntry newEntry = new SimpleEntry(true, true);
-    cache.put(newKey, newEntry);
-    assertNull(cache.getCurrentEntry(oldKey));
-    assertEquals(newEntry, cache.getCurrentEntry(newKey));
+    SimpleValue newValue = new SimpleValue(true, true);
+    cache.put(newKey, newValue);
+    assertNull(cache.getCurrentValue(oldKey));
+    assertEquals(newValue, cache.getCurrentValue(newKey));
     assertEquals(1, cache.size());
   }
 
   @Test
-  public void testOlderEntryIsIgnored() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+  public void testOlderValueIsIgnored() {
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(1);
 
-    SimpleEntry currentEntry = new SimpleEntry(true, true);
-    SimpleEntry notAsCurrentEntry = new SimpleEntry(true, false);
-    cache.put(DEFAULT_KEY, currentEntry);
-    cache.put(DEFAULT_KEY, notAsCurrentEntry);
+    SimpleValue currentValue = new SimpleValue(true, true);
+    SimpleValue notAsCurrentValue = new SimpleValue(true, false);
+    cache.put(DEFAULT_KEY, currentValue);
+    cache.put(DEFAULT_KEY, notAsCurrentValue);
     assertEquals(
-            "The existing entry in the cache was overwritten",
-            currentEntry,
-            cache.getCurrentEntry(DEFAULT_KEY)
+            "The existing value in the cache was overwritten",
+            currentValue,
+            cache.getCurrentValue(DEFAULT_KEY)
     );
   }
 
   @Test
-  public void testOutdatedEntryIsIgnored() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+  public void testOutdatedValueIsIgnored() {
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(1);
 
-    SimpleEntry outdatedEntry = new SimpleEntry(false, true);
-    cache.put(DEFAULT_KEY, outdatedEntry);
+    SimpleValue outdatedValue = new SimpleValue(false, true);
+    cache.put(DEFAULT_KEY, outdatedValue);
     assertEquals(0, cache.size());
-    assertNull(cache.getCurrentEntry(DEFAULT_KEY));
+    assertNull(cache.getCurrentValue(DEFAULT_KEY));
   }
 
   @Test
-  public void testCurrentEntryOverwritesExisting() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+  public void testCurrentValueOverwritesExisting() {
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(1);
 
-    SimpleEntry currentEntry = new SimpleEntry(true, true);
-    SimpleEntry notAsCurrentEntry = new SimpleEntry(true, false);
-    cache.put(DEFAULT_KEY, notAsCurrentEntry);
+    SimpleValue currentValue = new SimpleValue(true, true);
+    SimpleValue notAsCurrentValue = new SimpleValue(true, false);
+    cache.put(DEFAULT_KEY, notAsCurrentValue);
     assertEquals(1, cache.size());
-    cache.put(DEFAULT_KEY, currentEntry);
+    cache.put(DEFAULT_KEY, currentValue);
     assertEquals(1, cache.size());
     assertEquals(
-            "The existing entry in the cache was NOT overwritten",
-            currentEntry,
-            cache.getCurrentEntry(DEFAULT_KEY)
+            "The existing value in the cache was NOT overwritten",
+            currentValue,
+            cache.getCurrentValue(DEFAULT_KEY)
     );
   }
 
   @Test
-  public void testGetOutdatedEntryReturnsNull() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+  public void testGetOutdatedValueReturnsNull() {
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(1);
 
-    SimpleEntry entry = new SimpleEntry(true, true);
-    cache.put(DEFAULT_KEY, entry);
+    SimpleValue value = new SimpleValue(true, true);
+    cache.put(DEFAULT_KEY, value);
     assertEquals(1, cache.size());
-    assertEquals(entry, cache.getCurrentEntry(DEFAULT_KEY));
+    assertEquals(value, cache.getCurrentValue(DEFAULT_KEY));
 
-    entry.setCurrent(false);
-    assertNull("The entry should not be current anymore", cache.getCurrentEntry(DEFAULT_KEY));
+    value.setCurrent(false);
+    assertNull("The value should not be current anymore", cache.getCurrentValue(DEFAULT_KEY));
     assertEquals(0, cache.size());
   }
 
   @Test
   public void testRemove() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(1);
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(1);
 
-    SimpleEntry entry = new SimpleEntry(true, true);
-    cache.put(DEFAULT_KEY, entry);
+    SimpleValue value = new SimpleValue(true, true);
+    cache.put(DEFAULT_KEY, value);
     assertEquals(1, cache.size());
-    assertEquals(entry, cache.getCurrentEntry(DEFAULT_KEY));
+    assertEquals(value, cache.getCurrentValue(DEFAULT_KEY));
 
-    // remove the only entry
-    assertEquals(entry, cache.remove(DEFAULT_KEY));
-    assertNull(cache.getCurrentEntry(DEFAULT_KEY));
+    // remove the only value
+    assertEquals(value, cache.remove(DEFAULT_KEY));
+    assertNull(cache.getCurrentValue(DEFAULT_KEY));
     assertEquals(0, cache.size());
   }
 
   @Test
   public void testClear() {
-    LruCache<String, SimpleEntry> cache = new LruCache<String, SimpleEntry>(2);
+    LruCache<String, SimpleValue> cache = new LruCache<String, SimpleValue>(2);
 
     String key1 = DEFAULT_KEY + 1;
     String key2 = DEFAULT_KEY + 2;
-    SimpleEntry entry = new SimpleEntry(true, true);
-    cache.put(key1, entry);
-    cache.put(key2, entry);
-    assertEquals(entry, cache.getCurrentEntry(key1));
-    assertEquals(entry, cache.getCurrentEntry(key2));
+    SimpleValue value = new SimpleValue(true, true);
+    cache.put(key1, value);
+    cache.put(key2, value);
+    assertEquals(value, cache.getCurrentValue(key1));
+    assertEquals(value, cache.getCurrentValue(key2));
     assertEquals(2, cache.size());
 
     cache.clear();
-    assertNull(cache.getCurrentEntry(key1));
-    assertNull(cache.getCurrentEntry(key2));
+    assertNull(cache.getCurrentValue(key1));
+    assertNull(cache.getCurrentValue(key2));
     assertEquals(0, cache.size());
   }
 

From 250a3982519d40be1dae18ad6c047fa9ba49f283 Mon Sep 17 00:00:00 2001
From: Matthieu Martin <ma.tt.b.ma.rt.in+parquet@gmail.com>
Date: Mon, 23 Jun 2014 12:45:42 -0700
Subject: [PATCH 3/4] Be less aggressive about checking whether the underlying
 file has been appended to/overwritten/deleted in order to minimize the number
 of namenode interactions.

---
 .../main/java/parquet/hadoop/LruCache.java    | 10 +++----
 .../parquet/hadoop/ParquetInputFormat.java    | 28 ++++++-------------
 .../java/parquet/hadoop/TestInputFormat.java  | 14 ++--------
 .../java/parquet/hadoop/TestLruCache.java     |  4 +--
 4 files changed, 17 insertions(+), 39 deletions(-)

diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
index a8edc3d006..a4299869d5 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
@@ -14,7 +14,7 @@
  * @param <V> The value type.  Must extend {@link parquet.hadoop.LruCache.Value} so that the "staleness" of the value
  *           can be easily determined.
  */
-final class LruCache<K, V extends LruCache.Value<V>> {
+final class LruCache<K, V extends LruCache.Value<K, V>> {
   private static final Log LOG = Log.getLog(LruCache.class);
 
   private static final float DEFAULT_LOAD_FACTOR = 0.75f;
@@ -71,7 +71,7 @@ public V remove(K key) {
    * @param newValue value to be associated with the specified key
    */
   public void put(K key, V newValue) {
-    if (newValue == null || !newValue.isCurrent()) {
+    if (newValue == null || !newValue.isCurrent(key)) {
       if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because it is " +
               (newValue == null ? "null" : "not current"));
       return;
@@ -111,7 +111,7 @@ public void clear() {
   public V getCurrentValue(K key) {
     V value = cacheMap.get(key);
     if (Log.DEBUG) LOG.debug("Value for '" + key + "' " + (value == null ? "not " : "") + "in cache");
-    if (value != null && !value.isCurrent()) {
+    if (value != null && !value.isCurrent(key)) {
       // value is not current; remove it and return null
       remove(key);
       return null;
@@ -135,13 +135,13 @@ public int size() {
    *
    * @param <V>
    */
-  interface Value<V> {
+  interface Value<K, V> {
     /**
      * Is the value still current (e.g. has the referenced data been modified/updated in such a way that the value
      * is no longer useful)
      * @return {@code true} the value is still current, {@code false} the value is no longer useful
      */
-    public boolean isCurrent();
+    public boolean isCurrent(K key);
 
     /**
      * Compares this value with the specified value to check for relative age.
diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
index 73943440ad..4628097963 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
@@ -85,7 +85,7 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
 
   private static final int MIN_FOOTER_CACHE_SIZE = 100;
 
-  private LruCache<Path, FootersCacheValue> footersCache;
+  private LruCache<FileStatus, FootersCacheValue> footersCache;
 
   private Class<?> readSupportClass;
 
@@ -456,7 +456,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
 
     if (footersCache != null) {
       for (FileStatus status : statuses) {
-        FootersCacheValue cacheEntry = footersCache.getCurrentValue(status.getPath());
+        FootersCacheValue cacheEntry = footersCache.getCurrentValue(status);
         if (Log.DEBUG) LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "") + " found for '" + status.getPath() + "'");
         if (cacheEntry != null) {
           footers.add(cacheEntry.getFooter());
@@ -467,7 +467,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
     } else {
       // initialize the cache to store all of the current statuses or a default minimum size. The sizing of this LRU
       // cache was chosen to mimic prior behavior (i.e. so that performance would be at least as good as it was before)
-      footersCache = new LruCache<Path, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
+      footersCache = new LruCache<FileStatus, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
       missingStatuses.addAll(statuses);
     }
     if (Log.DEBUG) LOG.debug("found " + footers.size() + " footers in cache and adding up to " +
@@ -487,7 +487,7 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
       // Use the original file status objects to make sure we store a conservative (older) modification time (i.e. in
       // case the files and footers were modified and it's not clear which version of the footers we have)
       FileStatus fileStatus = missingStatusesMap.get(newFooter.getFile());
-      footersCache.put(fileStatus.getPath(), new FootersCacheValue(fileStatus, newFooter));
+      footersCache.put(fileStatus, new FootersCacheValue(fileStatus, newFooter));
     }
 
     footers.addAll(newFooters);
@@ -515,7 +515,7 @@ public GlobalMetaData getGlobalMetaData(JobContext jobContext) throws IOExceptio
     return ParquetFileWriter.getGlobalMetaData(getFooters(jobContext));
   }
 
-  static final class FootersCacheValue implements LruCache.Value<FootersCacheValue> {
+  static final class FootersCacheValue implements LruCache.Value<FileStatus, FootersCacheValue> {
     private final long modificationTime;
     private final Footer footer;
 
@@ -524,22 +524,10 @@ public FootersCacheValue(FileStatus status, Footer footer) {
       this.footer = new Footer(footer.getFile(), footer.getParquetMetadata());
     }
 
-    public boolean isCurrent() {
-      FileSystem fs;
-      FileStatus currentFile;
-      Path path = footer.getFile();
-      try {
-        fs = path.getFileSystem(new Configuration());
-        currentFile = fs.getFileStatus(path);
-      } catch (FileNotFoundException e) {
-        if (Log.DEBUG) LOG.debug("The '" + path + "' path was not found.");
-        return false;
-      } catch (IOException e) {
-        throw new RuntimeException("Exception while checking '" + path + "': " + e, e);
-      }
-      long currentModTime = currentFile.getModificationTime();
+    public boolean isCurrent(FileStatus key) {
+      long currentModTime = key.getModificationTime();
       boolean isCurrent = modificationTime >= currentModTime;
-      if (Log.DEBUG && !isCurrent) LOG.debug("The cache value for '" + currentFile.getPath() + "' is not current: " +
+      if (Log.DEBUG && !isCurrent) LOG.debug("The cache value for '" + key.getPath() + "' is not current: " +
               "cached modification time=" + modificationTime + ", current modification time: " + currentModTime);
       return isCurrent;
     }
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
index 67fa73a418..7a9ae4733f 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
@@ -253,17 +253,7 @@ public void testFooterCacheValueIsCurrent() throws IOException, InterruptedExcep
     ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
-    assertFalse(cacheValue.isCurrent());
-  }
-
-  @Test
-  public void testFooterCacheValueDeleted() throws IOException {
-    File tempFile = getTempFile();
-    FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
-
-    assertTrue(tempFile.delete());
-    assertFalse(cacheValue.isCurrent());
+    assertFalse(cacheValue.isCurrent(fs.getFileStatus(new Path(tempFile.getAbsolutePath()))));
   }
 
   @Test
@@ -293,7 +283,7 @@ private ParquetInputFormat.FootersCacheValue getDummyCacheValue(File file, FileS
     FileStatus status = fs.getFileStatus(path);
     ParquetMetadata mockMetadata = mock(ParquetMetadata.class);
     ParquetInputFormat.FootersCacheValue cacheValue = new ParquetInputFormat.FootersCacheValue(status, new Footer(path, mockMetadata));
-    assertTrue(cacheValue.isCurrent());
+    assertTrue(cacheValue.isCurrent(status));
     return cacheValue;
   }
 
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
index 7081d54f8f..91e24d8632 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestLruCache.java
@@ -7,7 +7,7 @@
 public class TestLruCache {
   private static final String DEFAULT_KEY = "test";
 
-  private static final class SimpleValue implements LruCache.Value<SimpleValue> {
+  private static final class SimpleValue implements LruCache.Value<String, SimpleValue> {
     private boolean current;
     private boolean newerThan;
 
@@ -17,7 +17,7 @@ public SimpleValue(boolean current, boolean newerThan) {
     }
 
     @Override
-    public boolean isCurrent() {
+    public boolean isCurrent(String key) {
       return current;
     }
 

From 99bb5a3acdf6c998f5d344bbeaeb446031f47df0 Mon Sep 17 00:00:00 2001
From: Matthieu Martin <ma.tt.b.ma.rt.in+parquet@gmail.com>
Date: Tue, 24 Jun 2014 11:47:11 -0700
Subject: [PATCH 4/4] Minor javadoc and whitespace changes. Also added the
 FileStatusWrapper class to ParquetInputFormat to make sure that the debugging
 log statements print out meaningful paths.

---
 .../main/java/parquet/hadoop/LruCache.java    | 102 +++++++++------
 .../parquet/hadoop/ParquetInputFormat.java    | 117 +++++++++++++-----
 .../java/parquet/hadoop/TestInputFormat.java  |   8 +-
 3 files changed, 153 insertions(+), 74 deletions(-)

diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
index a4299869d5..e9ecb37d2b 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/LruCache.java
@@ -6,13 +6,15 @@
 import java.util.Map;
 
 /**
- * A basic implementation of an LRU cache.  Besides evicting the least recently used entries (either based on insertion
- * or access order), this class also checks for "stale" entries as entries are inserted or retrieved (note "staleness"
- * is defined by the entries themselves (see {@link parquet.hadoop.LruCache.Value}.
+ * A basic implementation of an LRU cache.  Besides evicting the least recently
+ * used entries (either based on insertion or access order), this class also
+ * checks for "stale" entries as entries are inserted or retrieved (note
+ * "staleness" is defined by the entries themselves (see
+ * {@link parquet.hadoop.LruCache.Value}).
  *
  * @param <K> The key type. Acts as the key in a {@link java.util.LinkedHashMap}
- * @param <V> The value type.  Must extend {@link parquet.hadoop.LruCache.Value} so that the "staleness" of the value
- *           can be easily determined.
+ * @param <V> The value type.  Must extend {@link parquet.hadoop.LruCache.Value}
+ *           so that the "staleness" of the value can be easily determined.
  */
 final class LruCache<K, V extends LruCache.Value<K, V>> {
   private static final Log LOG = Log.getLog(LruCache.class);
@@ -25,7 +27,7 @@ final class LruCache<K, V extends LruCache.Value<K, V>> {
    * Constructs an access-order based LRU cache with {@code maxSize} entries.
    * @param maxSize The maximum number of entries to store in the cache.
    */
-  public LruCache(int maxSize) {
+  public LruCache(final int maxSize) {
     this(maxSize, DEFAULT_LOAD_FACTOR, true);
   }
 
@@ -34,16 +36,21 @@ public LruCache(int maxSize) {
    *
    * @param maxSize The maximum number of entries to store in the cache.
    * @param loadFactor Used to determine the initial capacity.
-   * @param accessOrder the ordering mode - {@code true} for access-order, {@code false} for insertion-order
+   * @param accessOrder the ordering mode - {@code true} for access-order,
+   * {@code false} for insertion-order
    */
-  public LruCache(final int maxSize, float loadFactor, boolean accessOrder) {
+  public LruCache(final int maxSize, final float loadFactor, final boolean accessOrder) {
+    int initialCapacity = Math.round(maxSize / loadFactor);
     cacheMap =
-            new LinkedHashMap<K, V>(Math.round(maxSize / loadFactor), loadFactor, accessOrder) {
+            new LinkedHashMap<K, V>(initialCapacity, loadFactor, accessOrder) {
               @Override
-              public boolean removeEldestEntry(Map.Entry<K,V> eldest) {
+              public boolean removeEldestEntry(final Map.Entry<K, V> eldest) {
                 boolean result = size() > maxSize;
                 if (result) {
-                  if (Log.DEBUG) LOG.debug("Removing eldest entry in cache: " + eldest.getKey());
+                  if (Log.DEBUG) {
+                    LOG.debug("Removing eldest entry in cache: "
+                            + eldest.getKey());
+                  }
                 }
                 return result;
               }
@@ -53,33 +60,42 @@ public boolean removeEldestEntry(Map.Entry<K,V> eldest) {
   /**
    * Removes the mapping for the specified key from this cache if present.
    * @param key key whose mapping is to be removed from the cache
-   * @return the previous value associated with key, or null if there was no mapping for key.
+   * @return the previous value associated with key, or null if there was no
+   * mapping for key.
    */
-  public V remove(K key) {
+  public V remove(final K key) {
     V oldValue = cacheMap.remove(key);
     if (oldValue != null) {
-      if (Log.DEBUG) LOG.debug("Removed cache entry for '" + key + "'");
+      if (Log.DEBUG) {
+        LOG.debug("Removed cache entry for '" + key + "'");
+      }
     }
     return oldValue;
   }
 
   /**
-   * Associates the specified value with the specified key in this cache. The value is only inserted if it is not null
-   * and it is considered current. If the cache previously contained a mapping for the key, the old value is replaced
-   * only if the new value is "newer" than the old one.
+   * Associates the specified value with the specified key in this cache. The
+   * value is only inserted if it is not null and it is considered current. If
+   * the cache previously contained a mapping for the key, the old value is
+   * replaced only if the new value is "newer" than the old one.
    * @param key key with which the specified value is to be associated
    * @param newValue value to be associated with the specified key
    */
-  public void put(K key, V newValue) {
+  public void put(final K key, final V newValue) {
     if (newValue == null || !newValue.isCurrent(key)) {
-      if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because it is " +
-              (newValue == null ? "null" : "not current"));
+      if (Log.WARN) {
+        LOG.warn("Ignoring new cache entry for '" + key + "' because it is "
+                + (newValue == null ? "null" : "not current"));
+      }
       return;
     }
 
     V oldValue = cacheMap.get(key);
     if (oldValue != null && oldValue.isNewerThan(newValue)) {
-      if (Log.WARN) LOG.warn("Ignoring new cache entry for '" + key + "' because existing cache entry is newer");
+      if (Log.WARN) {
+        LOG.warn("Ignoring new cache entry for '" + key + "' because "
+                + "existing cache entry is newer");
+      }
       return;
     }
 
@@ -95,22 +111,26 @@ public void put(K key, V newValue) {
   }
 
   /**
-   * Removes all of the mappings from this cache. The cache will be empty after this call returns.
+   * Removes all of the mappings from this cache. The cache will be empty
+   * after this call returns.
    */
   public void clear() {
     cacheMap.clear();
   }
 
   /**
-   * Returns the value to which the specified key is mapped, or null if 1) the value is not current or 2) this cache
-   * contains no mapping for the key.
+   * Returns the value to which the specified key is mapped, or null if 1) the
+   * value is not current or 2) this cache contains no mapping for the key.
    * @param key the key whose associated value is to be returned
-   * @return the value to which the specified key is mapped, or null if 1) the value is not current or 2) this cache
-   * contains no mapping for the key
+   * @return the value to which the specified key is mapped, or null if 1) the
+   * value is not current or 2) this cache contains no mapping for the key
    */
-  public V getCurrentValue(K key) {
+  public V getCurrentValue(final K key) {
     V value = cacheMap.get(key);
-    if (Log.DEBUG) LOG.debug("Value for '" + key + "' " + (value == null ? "not " : "") + "in cache");
+    if (Log.DEBUG) {
+      LOG.debug("Value for '" + key + "' " + (value == null ? "not " : "")
+              + "in cache");
+    }
     if (value != null && !value.isCurrent(key)) {
       // value is not current; remove it and return null
       remove(key);
@@ -129,27 +149,33 @@ public int size() {
   }
 
   /**
-   * {@link parquet.hadoop.LruCache} expects all values to follow this interface so the cache can determine
-   * 1) whether values are current (e.g. the referenced data has not been modified/updated in such a way that the value
-   * is no longer useful) and 2) whether a value is strictly "newer" than another value.
+   * {@link parquet.hadoop.LruCache} expects all values to follow this
+   * interface so the cache can determine 1) whether values are current (e.g.
+   * the referenced data has not been modified/updated in such a way that the
+   * value is no longer useful) and 2) whether a value is strictly "newer"
+   * than another value.
    *
-   * @param <V>
+   * @param <K> The key type.
+   * @param <V> Provides a bound for the {@link #isNewerThan(V)} method
    */
   interface Value<K, V> {
     /**
-     * Is the value still current (e.g. has the referenced data been modified/updated in such a way that the value
-     * is no longer useful)
-     * @return {@code true} the value is still current, {@code false} the value is no longer useful
+     * Is the value still current (e.g. has the referenced data been
+     * modified/updated in such a way that the value is no longer useful)
+     * @param key the key associated with this value
+     * @return {@code true} the value is still current, {@code false} the value
+     * is no longer useful
      */
-    public boolean isCurrent(K key);
+    boolean isCurrent(K key);
 
     /**
      * Compares this value with the specified value to check for relative age.
      * @param otherValue the value to be compared.
-     * @return {@code true} the value is strictly newer than the other value, {@code false} the value is older or just
+     * @return {@code true} the value is strictly newer than the other value,
+     * {@code false} the value is older or just
      * as new as the other value.
      */
-    public boolean isNewerThan(V otherValue);
+    boolean isNewerThan(V otherValue);
   }
 
 }
diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
index 4628097963..ef15542ed9 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
@@ -15,7 +15,6 @@
  */
 package parquet.hadoop;
 
-import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -85,7 +84,7 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
 
   private static final int MIN_FOOTER_CACHE_SIZE = 100;
 
-  private LruCache<FileStatus, FootersCacheValue> footersCache;
+  private LruCache<FileStatusWrapper, FootersCacheValue> footersCache;
 
   private Class<?> readSupportClass;
 
@@ -453,40 +452,46 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
     Configuration config = ContextUtil.getConfiguration(jobContext);
     List<Footer> footers = new ArrayList<Footer>(statuses.size());
     Set<FileStatus> missingStatuses = new HashSet<FileStatus>();
-
-    if (footersCache != null) {
-      for (FileStatus status : statuses) {
-        FootersCacheValue cacheEntry = footersCache.getCurrentValue(status);
-        if (Log.DEBUG) LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "") + " found for '" + status.getPath() + "'");
-        if (cacheEntry != null) {
-          footers.add(cacheEntry.getFooter());
-        } else {
-          missingStatuses.add(status);
-        }
+    Map<Path, FileStatusWrapper> missingStatusesMap =
+            new HashMap<Path, FileStatusWrapper>(missingStatuses.size());
+
+    if (footersCache == null) {
+      footersCache =
+              new LruCache<FileStatusWrapper, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
+    }
+    for (FileStatus status : statuses) {
+      FileStatusWrapper statusWrapper = new FileStatusWrapper(status);
+      FootersCacheValue cacheEntry =
+              footersCache.getCurrentValue(statusWrapper);
+      if (Log.DEBUG) {
+        LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "")
+                + " found for '" + status.getPath() + "'");
       }
-    } else {
-      // initialize the cache to store all of the current statuses or a default minimum size. The sizing of this LRU
-      // cache was chosen to mimic prior behavior (i.e. so that performance would be at least as good as it was before)
-      footersCache = new LruCache<FileStatus, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
-      missingStatuses.addAll(statuses);
+      if (cacheEntry != null) {
+        footers.add(cacheEntry.getFooter());
+      } else {
+        missingStatuses.add(status);
+        missingStatusesMap.put(status.getPath(), statusWrapper);
+      }
+    }
+    if (Log.DEBUG) {
+      LOG.debug("found " + footers.size() + " footers in cache and adding up "
+              + "to " + missingStatuses.size() + " missing footers to the cache");
     }
-    if (Log.DEBUG) LOG.debug("found " + footers.size() + " footers in cache and adding up to " +
-            missingStatuses.size() + " missing footers to the cache");
 
 
     if (missingStatuses.isEmpty()) {
       return footers;
     }
 
-    List<Footer> newFooters = getFooters(config, new ArrayList<FileStatus>(missingStatuses));
-    Map<Path, FileStatus> missingStatusesMap = new HashMap<Path, FileStatus>(missingStatuses.size());
-    for (FileStatus missingStatus : missingStatuses) {
-      missingStatusesMap.put(missingStatus.getPath(), missingStatus);
-    }
+    List<Footer> newFooters =
+            getFooters(config, new ArrayList<FileStatus>(missingStatuses));
     for (Footer newFooter : newFooters) {
-      // Use the original file status objects to make sure we store a conservative (older) modification time (i.e. in
-      // case the files and footers were modified and it's not clear which version of the footers we have)
-      FileStatus fileStatus = missingStatusesMap.get(newFooter.getFile());
+      // Use the original file status objects to make sure we store a
+      // conservative (older) modification time (i.e. in case the files and
+      // footers were modified and it's not clear which version of the footers
+      // we have)
+      FileStatusWrapper fileStatus = missingStatusesMap.get(newFooter.getFile());
       footersCache.put(fileStatus, new FootersCacheValue(fileStatus, newFooter));
     }
 
@@ -515,20 +520,30 @@ public GlobalMetaData getGlobalMetaData(JobContext jobContext) throws IOExceptio
     return ParquetFileWriter.getGlobalMetaData(getFooters(jobContext));
   }
 
-  static final class FootersCacheValue implements LruCache.Value<FileStatus, FootersCacheValue> {
+  /**
+   * A simple wrapper around {@link parquet.hadoop.Footer} that also includes a
+   * modification time associated with that footer.  The modification time is
+   * used to determine whether the footer is still current.
+   */
+  static final class FootersCacheValue
+          implements LruCache.Value<FileStatusWrapper, FootersCacheValue> {
     private final long modificationTime;
     private final Footer footer;
 
-    public FootersCacheValue(FileStatus status, Footer footer) {
+    public FootersCacheValue(FileStatusWrapper status, Footer footer) {
       this.modificationTime = status.getModificationTime();
       this.footer = new Footer(footer.getFile(), footer.getParquetMetadata());
     }
 
-    public boolean isCurrent(FileStatus key) {
+    @Override
+    public boolean isCurrent(FileStatusWrapper key) {
       long currentModTime = key.getModificationTime();
       boolean isCurrent = modificationTime >= currentModTime;
-      if (Log.DEBUG && !isCurrent) LOG.debug("The cache value for '" + key.getPath() + "' is not current: " +
-              "cached modification time=" + modificationTime + ", current modification time: " + currentModTime);
+      if (Log.DEBUG && !isCurrent) {
+        LOG.debug("The cache value for '" + key + "' is not current: "
+                + "cached modification time=" + modificationTime + ", "
+                + "current modification time: " + currentModTime);
+      }
       return isCurrent;
     }
 
@@ -536,8 +551,10 @@ public Footer getFooter() {
       return footer;
     }
 
+    @Override
     public boolean isNewerThan(FootersCacheValue otherValue) {
-      return otherValue == null || modificationTime > otherValue.modificationTime;
+      return otherValue == null ||
+              modificationTime > otherValue.modificationTime;
     }
 
     public Path getPath() {
@@ -545,4 +562,38 @@ public Path getPath() {
     }
   }
 
+  /**
+   * A simple wrapper around {@link org.apache.hadoop.fs.FileStatus} with a
+   * meaningful "toString()" method
+   */
+  static final class FileStatusWrapper {
+    private final FileStatus status;
+    public FileStatusWrapper(FileStatus fileStatus) {
+      if (fileStatus == null) {
+        throw new IllegalArgumentException("FileStatus object cannot be null");
+      }
+      status = fileStatus;
+    }
+
+    public long getModificationTime() {
+      return status.getModificationTime();
+    }
+
+    @Override
+    public int hashCode() {
+      return status.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object other) {
+      return other instanceof FileStatusWrapper &&
+              status.equals(((FileStatusWrapper) other).status);
+    }
+
+    @Override
+    public String toString() {
+      return status.getPath().toString();
+    }
+  }
+
 }
diff --git a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
index 7a9ae4733f..1ab1cc50b1 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
@@ -253,7 +253,7 @@ public void testFooterCacheValueIsCurrent() throws IOException, InterruptedExcep
     ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
-    assertFalse(cacheValue.isCurrent(fs.getFileStatus(new Path(tempFile.getAbsolutePath()))));
+    assertFalse(cacheValue.isCurrent(new ParquetInputFormat.FileStatusWrapper(fs.getFileStatus(new Path(tempFile.getAbsolutePath())))));
   }
 
   @Test
@@ -281,9 +281,11 @@ private File getTempFile() throws IOException {
   private ParquetInputFormat.FootersCacheValue getDummyCacheValue(File file, FileSystem fs) throws IOException {
     Path path = new Path(file.getPath());
     FileStatus status = fs.getFileStatus(path);
+    ParquetInputFormat.FileStatusWrapper statusWrapper = new ParquetInputFormat.FileStatusWrapper(status);
     ParquetMetadata mockMetadata = mock(ParquetMetadata.class);
-    ParquetInputFormat.FootersCacheValue cacheValue = new ParquetInputFormat.FootersCacheValue(status, new Footer(path, mockMetadata));
-    assertTrue(cacheValue.isCurrent(status));
+    ParquetInputFormat.FootersCacheValue cacheValue =
+            new ParquetInputFormat.FootersCacheValue(statusWrapper, new Footer(path, mockMetadata));
+    assertTrue(cacheValue.isCurrent(statusWrapper));
     return cacheValue;
   }