From bc043d55221f863cfc71cdb70683848b23f3096b Mon Sep 17 00:00:00 2001
From: Cheng Lian <lian.cs.zju@gmail.com>
Date: Wed, 30 Jul 2014 12:59:30 +0800
Subject: [PATCH 1/2] Leveraged footersCache to pass FileStatus objects without
 side effects

---
 .../parquet/hadoop/ParquetInputFormat.java    | 117 +++++++++++-------
 .../java/parquet/hadoop/TestInputFormat.java  |  12 +-
 2 files changed, 76 insertions(+), 53 deletions(-)

diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
index 0231ccd720..b364a97365 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
@@ -89,7 +89,7 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
    * key to configure the filter
    */
   public static final String UNBOUND_RECORD_FILTER = "parquet.read.filter";
-  
+
   /**
    * key to configure type checking for conflicting schemas (default: true)
    */
@@ -102,7 +102,7 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
 
   private static final int MIN_FOOTER_CACHE_SIZE = 100;
 
-  private LruCache<FileStatusWrapper, FootersCacheValue> footersCache;
+  private LruCache<FileStatusWrapper, CachedFooter> footersCache;
 
   private Class<?> readSupportClass;
 
@@ -414,23 +414,21 @@ private static void checkSorted(List<BlockMetaData> rowGroupBlocks) {
   @Override
   public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
     List<InputSplit> splits = new ArrayList<InputSplit>();
-    splits.addAll(getSplits(ContextUtil.getConfiguration(jobContext), getFooters(jobContext)));
+    splits.addAll(getSplitsWithCachedFooters(ContextUtil.getConfiguration(jobContext), getCachedFooters(jobContext)));
     return splits;
   }
 
-  /**
-   * @param configuration the configuration to connect to the file system
-   * @param footers the footers of the files to read
-   * @return the splits for the footers
-   * @throws IOException
-   */
-  public List<ParquetInputSplit> getSplits(Configuration configuration, List<Footer> footers) throws IOException {
+  private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configuration, List<CachedFooter> cachedFooters) throws IOException {
     final long maxSplitSize = configuration.getLong("mapred.max.split.size", Long.MAX_VALUE);
     final long minSplitSize = Math.max(getFormatMinSplitSize(), configuration.getLong("mapred.min.split.size", 0L));
     if (maxSplitSize < 0 || minSplitSize < 0) {
       throw new ParquetDecodingException("maxSplitSize or minSplitSie should not be negative: maxSplitSize = " + maxSplitSize + "; minSplitSize = " + minSplitSize);
     }
     List<ParquetInputSplit> splits = new ArrayList<ParquetInputSplit>();
+    List<Footer> footers = new ArrayList<Footer>(cachedFooters.size());
+    for (CachedFooter cacheFooter : cachedFooters) {
+      footers.add(cacheFooter.footer);
+    }
     GlobalMetaData globalMetaData = ParquetFileWriter.getGlobalMetaData(footers, configuration.getBoolean(STRICT_TYPE_CHECKING, true));
     ReadContext readContext = getReadSupport(configuration).init(new InitContext(
         configuration,
@@ -442,18 +440,15 @@ public List<ParquetInputSplit> getSplits(Configuration configuration, List<Foote
     long rowGroupsDropped = 0;
     long totalRowGroups = 0;
 
-    for (Footer footer : footers) {
-      final Path file = footer.getFile();
+    for (CachedFooter cachedFooter : cachedFooters) {
+      final Path file = cachedFooter.footer.getFile();
       LOG.debug(file);
       FileSystem fs = file.getFileSystem(configuration);
-      FileStatus fileStatus = fs.getFileStatus(file);
-      ParquetMetadata parquetMetaData = footer.getParquetMetadata();
+      FileStatus fileStatus = cachedFooter.getFileStatus();
+      ParquetMetadata parquetMetaData = cachedFooter.footer.getParquetMetadata();
       List<BlockMetaData> blocks = parquetMetaData.getBlocks();
-
-      List<BlockMetaData> filteredBlocks = blocks;
-
       totalRowGroups += blocks.size();
-      filteredBlocks = RowGroupFilter.filterRowGroups(filter, blocks, parquetMetaData.getFileMetaData().getSchema());
+      List<BlockMetaData> filteredBlocks = RowGroupFilter.filterRowGroups(filter, blocks, parquetMetaData.getFileMetaData().getSchema());
       rowGroupsDropped += blocks.size() - filteredBlocks.size();
 
       if (filteredBlocks.isEmpty()) {
@@ -470,8 +465,7 @@ public List<ParquetInputSplit> getSplits(Configuration configuration, List<Foote
               readContext.getRequestedSchema().toString(),
               readContext.getReadSupportMetadata(),
               minSplitSize,
-              maxSplitSize)
-          );
+              maxSplitSize));
     }
 
     if (rowGroupsDropped > 0 && totalRowGroups > 0) {
@@ -484,6 +478,24 @@ public List<ParquetInputSplit> getSplits(Configuration configuration, List<Foote
     return splits;
   }
 
+  /**
+   * @param configuration the configuration to connect to the file system
+   * @param footers the footers of the files to read
+   * @return the splits for the footers
+   * @throws IOException
+   */
+  public List<ParquetInputSplit> getSplits(Configuration configuration, List<Footer> footers) throws IOException {
+    List<CachedFooter> cachedFooters = new ArrayList<CachedFooter>(footers.size());
+
+    for (Footer footer : footers) {
+      FileSystem fs = footer.getFile().getFileSystem(configuration);
+      FileStatus status = fs.getFileStatus(footer.getFile());
+      cachedFooters.add(new CachedFooter(new FileStatusWrapper(status), footer));
+    }
+
+    return getSplitsWithCachedFooters(configuration, cachedFooters);
+  }
+
   /*
    * This is to support multi-level/recursive directory listing until
    * MAPREDUCE-1577 is fixed.
@@ -491,7 +503,7 @@ public List<ParquetInputSplit> getSplits(Configuration configuration, List<Foote
   @Override
   protected List<FileStatus> listStatus(JobContext jobContext) throws IOException {
     return getAllFileRecursively(super.listStatus(jobContext),
-       ContextUtil.getConfiguration(jobContext));
+        ContextUtil.getConfiguration(jobContext));
   }
 
   private static List<FileStatus> getAllFileRecursively(
@@ -529,37 +541,28 @@ public boolean accept(Path p){
     }
   };
 
-  /**
-   * @param jobContext the current job context
-   * @return the footers for the files
-   * @throws IOException
-   */
-  public List<Footer> getFooters(JobContext jobContext) throws IOException {
+  private List<CachedFooter> getCachedFooters(JobContext jobContext) throws IOException {
     List<FileStatus> statuses = listStatus(jobContext);
     if (statuses.isEmpty()) {
       return Collections.emptyList();
     }
 
     Configuration config = ContextUtil.getConfiguration(jobContext);
-    List<Footer> footers = new ArrayList<Footer>(statuses.size());
+    List<CachedFooter> footers = new ArrayList<CachedFooter>(statuses.size());
     Set<FileStatus> missingStatuses = new HashSet<FileStatus>();
-    Map<Path, FileStatusWrapper> missingStatusesMap =
-            new HashMap<Path, FileStatusWrapper>(missingStatuses.size());
+    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));
+      footersCache = new LruCache<FileStatusWrapper, CachedFooter>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
     }
     for (FileStatus status : statuses) {
       FileStatusWrapper statusWrapper = new FileStatusWrapper(status);
-      FootersCacheValue cacheEntry =
-              footersCache.getCurrentValue(statusWrapper);
+      CachedFooter cacheEntry = footersCache.getCurrentValue(statusWrapper);
       if (Log.DEBUG) {
-        LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "")
-                + " found for '" + status.getPath() + "'");
+        LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "") + " found for '" + status.getPath() + "'");
       }
       if (cacheEntry != null) {
-        footers.add(cacheEntry.getFooter());
+        footers.add(cacheEntry);
       } else {
         missingStatuses.add(status);
         missingStatusesMap.put(status.getPath(), statusWrapper);
@@ -575,18 +578,34 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
       return footers;
     }
 
-    List<Footer> newFooters =
-            getFooters(config, new ArrayList<FileStatus>(missingStatuses));
+    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)
       FileStatusWrapper fileStatus = missingStatusesMap.get(newFooter.getFile());
-      footersCache.put(fileStatus, new FootersCacheValue(fileStatus, newFooter));
+      CachedFooter cacheEntry = new CachedFooter(fileStatus, newFooter);
+      footersCache.put(fileStatus, cacheEntry);
+      footers.add(cacheEntry);
+    }
+
+    return footers;
+  }
+
+  /**
+   * @param jobContext the current job context
+   * @return the footers for the files
+   * @throws IOException
+   */
+  public List<Footer> getFooters(JobContext jobContext) throws IOException {
+    List<CachedFooter> cacheFooters= getCachedFooters(jobContext);
+    List<Footer> footers = new ArrayList<Footer>(cacheFooters.size());
+
+    for (CachedFooter cachedFooter: cacheFooters) {
+      footers.add(cachedFooter.footer);
     }
 
-    footers.addAll(newFooters);
     return footers;
   }
 
@@ -616,14 +635,15 @@ public GlobalMetaData getGlobalMetaData(JobContext jobContext) throws IOExceptio
    * 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> {
+  static final class CachedFooter implements LruCache.Value<FileStatusWrapper, CachedFooter> {
     private final long modificationTime;
     private final Footer footer;
+    private final FileStatus status;
 
-    public FootersCacheValue(FileStatusWrapper status, Footer footer) {
+    public CachedFooter(FileStatusWrapper status, Footer footer) {
       this.modificationTime = status.getModificationTime();
       this.footer = new Footer(footer.getFile(), footer.getParquetMetadata());
+      this.status = status.status;
     }
 
     @Override
@@ -642,10 +662,13 @@ public Footer getFooter() {
       return footer;
     }
 
+    public FileStatus getFileStatus() {
+      return status;
+    }
+
     @Override
-    public boolean isNewerThan(FootersCacheValue otherValue) {
-      return otherValue == null ||
-              modificationTime > otherValue.modificationTime;
+    public boolean isNewerThan(CachedFooter 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 1823716e97..e7fc6693e8 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
@@ -340,7 +340,7 @@ public static BlockMetaData makeBlockFromStats(IntStatistics stats, long valueCo
   public void testFooterCacheValueIsCurrent() throws IOException, InterruptedException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
+    ParquetInputFormat.CachedFooter cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
     assertFalse(cacheValue.isCurrent(new ParquetInputFormat.FileStatusWrapper(fs.getFileStatus(new Path(tempFile.getAbsolutePath())))));
@@ -350,13 +350,13 @@ public void testFooterCacheValueIsCurrent() throws IOException, InterruptedExcep
   public void testFooterCacheValueIsNewer() throws IOException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
+    ParquetInputFormat.CachedFooter cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(cacheValue.isNewerThan(null));
     assertFalse(cacheValue.isNewerThan(cacheValue));
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
-    ParquetInputFormat.FootersCacheValue newerCacheValue = getDummyCacheValue(tempFile, fs);
+    ParquetInputFormat.CachedFooter newerCacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(newerCacheValue.isNewerThan(cacheValue));
     assertFalse(cacheValue.isNewerThan(newerCacheValue));
@@ -368,13 +368,13 @@ private File getTempFile() throws IOException {
     return tempFile;
   }
 
-  private ParquetInputFormat.FootersCacheValue getDummyCacheValue(File file, FileSystem fs) throws IOException {
+  private ParquetInputFormat.CachedFooter 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(statusWrapper, new Footer(path, mockMetadata));
+    ParquetInputFormat.CachedFooter cacheValue =
+            new ParquetInputFormat.CachedFooter(statusWrapper, new Footer(path, mockMetadata));
     assertTrue(cacheValue.isCurrent(statusWrapper));
     return cacheValue;
   }

From c4f2fb8b6f29c8062b2eda407447a7d027b80612 Mon Sep 17 00:00:00 2001
From: Cheng Lian <lian.cs.zju@gmail.com>
Date: Tue, 26 Aug 2014 18:25:06 -0700
Subject: [PATCH 2/2] Reverted code style changes to make this PR cleaner

---
 .../parquet/hadoop/ParquetInputFormat.java    | 85 +++++++++++--------
 .../java/parquet/hadoop/TestInputFormat.java  | 12 +--
 2 files changed, 54 insertions(+), 43 deletions(-)

diff --git a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
index b364a97365..c3f7d81cbb 100644
--- a/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/parquet/hadoop/ParquetInputFormat.java
@@ -102,7 +102,7 @@ public class ParquetInputFormat<T> extends FileInputFormat<Void, T> {
 
   private static final int MIN_FOOTER_CACHE_SIZE = 100;
 
-  private LruCache<FileStatusWrapper, CachedFooter> footersCache;
+  private LruCache<FileStatusWrapper, FootersCacheValue> footersCache;
 
   private Class<?> readSupportClass;
 
@@ -418,7 +418,7 @@ public List<InputSplit> getSplits(JobContext jobContext) throws IOException {
     return splits;
   }
 
-  private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configuration, List<CachedFooter> cachedFooters) throws IOException {
+  private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configuration, List<FootersCacheValue> cachedFooters) throws IOException {
     final long maxSplitSize = configuration.getLong("mapred.max.split.size", Long.MAX_VALUE);
     final long minSplitSize = Math.max(getFormatMinSplitSize(), configuration.getLong("mapred.min.split.size", 0L));
     if (maxSplitSize < 0 || minSplitSize < 0) {
@@ -426,7 +426,7 @@ private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configu
     }
     List<ParquetInputSplit> splits = new ArrayList<ParquetInputSplit>();
     List<Footer> footers = new ArrayList<Footer>(cachedFooters.size());
-    for (CachedFooter cacheFooter : cachedFooters) {
+    for (FootersCacheValue cacheFooter : cachedFooters) {
       footers.add(cacheFooter.footer);
     }
     GlobalMetaData globalMetaData = ParquetFileWriter.getGlobalMetaData(footers, configuration.getBoolean(STRICT_TYPE_CHECKING, true));
@@ -440,15 +440,18 @@ private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configu
     long rowGroupsDropped = 0;
     long totalRowGroups = 0;
 
-    for (CachedFooter cachedFooter : cachedFooters) {
+    for (FootersCacheValue cachedFooter : cachedFooters) {
       final Path file = cachedFooter.footer.getFile();
       LOG.debug(file);
       FileSystem fs = file.getFileSystem(configuration);
       FileStatus fileStatus = cachedFooter.getFileStatus();
       ParquetMetadata parquetMetaData = cachedFooter.footer.getParquetMetadata();
       List<BlockMetaData> blocks = parquetMetaData.getBlocks();
+
+      List<BlockMetaData> filteredBlocks = blocks;
+
       totalRowGroups += blocks.size();
-      List<BlockMetaData> filteredBlocks = RowGroupFilter.filterRowGroups(filter, blocks, parquetMetaData.getFileMetaData().getSchema());
+      filteredBlocks = RowGroupFilter.filterRowGroups(filter, blocks, parquetMetaData.getFileMetaData().getSchema());
       rowGroupsDropped += blocks.size() - filteredBlocks.size();
 
       if (filteredBlocks.isEmpty()) {
@@ -465,7 +468,8 @@ private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configu
               readContext.getRequestedSchema().toString(),
               readContext.getReadSupportMetadata(),
               minSplitSize,
-              maxSplitSize));
+              maxSplitSize)
+          );
     }
 
     if (rowGroupsDropped > 0 && totalRowGroups > 0) {
@@ -485,12 +489,12 @@ private List<ParquetInputSplit> getSplitsWithCachedFooters(Configuration configu
    * @throws IOException
    */
   public List<ParquetInputSplit> getSplits(Configuration configuration, List<Footer> footers) throws IOException {
-    List<CachedFooter> cachedFooters = new ArrayList<CachedFooter>(footers.size());
+    List<FootersCacheValue> cachedFooters = new ArrayList<FootersCacheValue>(footers.size());
 
     for (Footer footer : footers) {
       FileSystem fs = footer.getFile().getFileSystem(configuration);
       FileStatus status = fs.getFileStatus(footer.getFile());
-      cachedFooters.add(new CachedFooter(new FileStatusWrapper(status), footer));
+      cachedFooters.add(new FootersCacheValue(new FileStatusWrapper(status), footer));
     }
 
     return getSplitsWithCachedFooters(configuration, cachedFooters);
@@ -503,7 +507,7 @@ public List<ParquetInputSplit> getSplits(Configuration configuration, List<Foote
   @Override
   protected List<FileStatus> listStatus(JobContext jobContext) throws IOException {
     return getAllFileRecursively(super.listStatus(jobContext),
-        ContextUtil.getConfiguration(jobContext));
+       ContextUtil.getConfiguration(jobContext));
   }
 
   private static List<FileStatus> getAllFileRecursively(
@@ -541,25 +545,45 @@ public boolean accept(Path p){
     }
   };
 
-  private List<CachedFooter> getCachedFooters(JobContext jobContext) throws IOException {
+  /**
+   * @param jobContext the current job context
+   * @return the footers for the files
+   * @throws IOException
+   */
+  public List<Footer> getFooters(JobContext jobContext) throws IOException {
+    List<FootersCacheValue> cacheFooters= getCachedFooters(jobContext);
+    List<Footer> footers = new ArrayList<Footer>(cacheFooters.size());
+
+    for (FootersCacheValue cachedFooter: cacheFooters) {
+      footers.add(cachedFooter.footer);
+    }
+
+    return footers;
+  }
+
+  private List<FootersCacheValue> getCachedFooters(JobContext jobContext) throws IOException {
     List<FileStatus> statuses = listStatus(jobContext);
     if (statuses.isEmpty()) {
       return Collections.emptyList();
     }
 
     Configuration config = ContextUtil.getConfiguration(jobContext);
-    List<CachedFooter> footers = new ArrayList<CachedFooter>(statuses.size());
+    List<FootersCacheValue> footers = new ArrayList<FootersCacheValue>(statuses.size());
     Set<FileStatus> missingStatuses = new HashSet<FileStatus>();
-    Map<Path, FileStatusWrapper> missingStatusesMap = new HashMap<Path, FileStatusWrapper>(missingStatuses.size());
+    Map<Path, FileStatusWrapper> missingStatusesMap =
+            new HashMap<Path, FileStatusWrapper>(missingStatuses.size());
 
     if (footersCache == null) {
-      footersCache = new LruCache<FileStatusWrapper, CachedFooter>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
+      footersCache =
+              new LruCache<FileStatusWrapper, FootersCacheValue>(Math.max(statuses.size(), MIN_FOOTER_CACHE_SIZE));
     }
     for (FileStatus status : statuses) {
       FileStatusWrapper statusWrapper = new FileStatusWrapper(status);
-      CachedFooter cacheEntry = footersCache.getCurrentValue(statusWrapper);
+      FootersCacheValue cacheEntry =
+              footersCache.getCurrentValue(statusWrapper);
       if (Log.DEBUG) {
-        LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "") + " found for '" + status.getPath() + "'");
+        LOG.debug("Cache entry " + (cacheEntry == null ? "not " : "")
+                + " found for '" + status.getPath() + "'");
       }
       if (cacheEntry != null) {
         footers.add(cacheEntry);
@@ -578,14 +602,15 @@ private List<CachedFooter> getCachedFooters(JobContext jobContext) throws IOExce
       return footers;
     }
 
-    List<Footer> newFooters = getFooters(config, new ArrayList<FileStatus>(missingStatuses));
+    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)
       FileStatusWrapper fileStatus = missingStatusesMap.get(newFooter.getFile());
-      CachedFooter cacheEntry = new CachedFooter(fileStatus, newFooter);
+      FootersCacheValue cacheEntry = new FootersCacheValue(fileStatus, newFooter);
       footersCache.put(fileStatus, cacheEntry);
       footers.add(cacheEntry);
     }
@@ -593,22 +618,6 @@ private List<CachedFooter> getCachedFooters(JobContext jobContext) throws IOExce
     return footers;
   }
 
-  /**
-   * @param jobContext the current job context
-   * @return the footers for the files
-   * @throws IOException
-   */
-  public List<Footer> getFooters(JobContext jobContext) throws IOException {
-    List<CachedFooter> cacheFooters= getCachedFooters(jobContext);
-    List<Footer> footers = new ArrayList<Footer>(cacheFooters.size());
-
-    for (CachedFooter cachedFooter: cacheFooters) {
-      footers.add(cachedFooter.footer);
-    }
-
-    return footers;
-  }
-
   /**
    * the footers for the files
    * @param configuration to connect to the file system
@@ -635,12 +644,13 @@ public GlobalMetaData getGlobalMetaData(JobContext jobContext) throws IOExceptio
    * modification time associated with that footer.  The modification time is
    * used to determine whether the footer is still current.
    */
-  static final class CachedFooter implements LruCache.Value<FileStatusWrapper, CachedFooter> {
+  static final class FootersCacheValue
+          implements LruCache.Value<FileStatusWrapper, FootersCacheValue> {
     private final long modificationTime;
     private final Footer footer;
     private final FileStatus status;
 
-    public CachedFooter(FileStatusWrapper status, Footer footer) {
+    public FootersCacheValue(FileStatusWrapper status, Footer footer) {
       this.modificationTime = status.getModificationTime();
       this.footer = new Footer(footer.getFile(), footer.getParquetMetadata());
       this.status = status.status;
@@ -667,8 +677,9 @@ public FileStatus getFileStatus() {
     }
 
     @Override
-    public boolean isNewerThan(CachedFooter otherValue) {
-      return otherValue == null || modificationTime > otherValue.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 e7fc6693e8..1823716e97 100644
--- a/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
+++ b/parquet-hadoop/src/test/java/parquet/hadoop/TestInputFormat.java
@@ -340,7 +340,7 @@ public static BlockMetaData makeBlockFromStats(IntStatistics stats, long valueCo
   public void testFooterCacheValueIsCurrent() throws IOException, InterruptedException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.CachedFooter cacheValue = getDummyCacheValue(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
     assertFalse(cacheValue.isCurrent(new ParquetInputFormat.FileStatusWrapper(fs.getFileStatus(new Path(tempFile.getAbsolutePath())))));
@@ -350,13 +350,13 @@ public void testFooterCacheValueIsCurrent() throws IOException, InterruptedExcep
   public void testFooterCacheValueIsNewer() throws IOException {
     File tempFile = getTempFile();
     FileSystem fs = FileSystem.getLocal(new Configuration());
-    ParquetInputFormat.CachedFooter cacheValue = getDummyCacheValue(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue cacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(cacheValue.isNewerThan(null));
     assertFalse(cacheValue.isNewerThan(cacheValue));
 
     assertTrue(tempFile.setLastModified(tempFile.lastModified() + 5000));
-    ParquetInputFormat.CachedFooter newerCacheValue = getDummyCacheValue(tempFile, fs);
+    ParquetInputFormat.FootersCacheValue newerCacheValue = getDummyCacheValue(tempFile, fs);
 
     assertTrue(newerCacheValue.isNewerThan(cacheValue));
     assertFalse(cacheValue.isNewerThan(newerCacheValue));
@@ -368,13 +368,13 @@ private File getTempFile() throws IOException {
     return tempFile;
   }
 
-  private ParquetInputFormat.CachedFooter getDummyCacheValue(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);
     ParquetInputFormat.FileStatusWrapper statusWrapper = new ParquetInputFormat.FileStatusWrapper(status);
     ParquetMetadata mockMetadata = mock(ParquetMetadata.class);
-    ParquetInputFormat.CachedFooter cacheValue =
-            new ParquetInputFormat.CachedFooter(statusWrapper, new Footer(path, mockMetadata));
+    ParquetInputFormat.FootersCacheValue cacheValue =
+            new ParquetInputFormat.FootersCacheValue(statusWrapper, new Footer(path, mockMetadata));
     assertTrue(cacheValue.isCurrent(statusWrapper));
     return cacheValue;
   }