From 5aac156adc17173844b9c22ca0734a4c6cecabd3 Mon Sep 17 00:00:00 2001
From: Aaron Niskode-Dossett <aniskodedossett@etsy.com>
Date: Fri, 17 Jul 2026 15:28:14 -0500
Subject: [PATCH] GH-3394: Reuse listed file status during split planning

---
 .../parquet/hadoop/ParquetInputFormat.java    |  44 ++-
 .../TestInputFormatFileStatusReuse.java       | 295 ++++++++++++++++++
 2 files changed, 333 insertions(+), 6 deletions(-)
 create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatFileStatusReuse.java

diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java
index 8e05d49bd3..811e1f3833 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java
@@ -485,15 +485,42 @@ public List<Footer> getFooters(JobContext jobContext) throws IOException {
 
       if (footer == null) {
         // Footer was originally missing, so get it from the cache again
-        footers.add(footersCache.getCurrentValue(footerEntry.getKey()).getFooter());
-      } else {
-        footers.add(footer);
+        footer = footersCache.getCurrentValue(footerEntry.getKey()).getFooter();
       }
+      footers.add(new FooterWithFileStatus(footer, footerEntry.getKey().status));
     }
 
     return footers;
   }
 
+  static FileStatus getFileStatus(Footer footer) {
+    return footer instanceof FooterWithFileStatus ? ((FooterWithFileStatus) footer).fileStatus : null;
+  }
+
+  private static List<Footer> withFileStatuses(List<Footer> footers, Collection<FileStatus> statuses) {
+    Map<Path, FileStatus> statusesByPath = new HashMap<Path, FileStatus>(statuses.size());
+    for (FileStatus status : statuses) {
+      statusesByPath.put(status.getPath(), status);
+    }
+
+    List<Footer> footersWithStatuses = new ArrayList<Footer>(footers.size());
+    for (Footer footer : footers) {
+      FileStatus status = statusesByPath.get(footer.getFile());
+      footersWithStatuses.add(status == null ? footer : new FooterWithFileStatus(footer, status));
+    }
+    return footersWithStatuses;
+  }
+
+  /** Carries an already-listed status through split planning without adding it to Footer's public API. */
+  private static final class FooterWithFileStatus extends Footer {
+    private final FileStatus fileStatus;
+
+    private FooterWithFileStatus(Footer footer, FileStatus fileStatus) {
+      super(footer.getFile(), footer.getParquetMetadata());
+      this.fileStatus = fileStatus;
+    }
+  }
+
   public List<Footer> getFooters(Configuration configuration, List<FileStatus> statuses) throws IOException {
     return getFooters(configuration, (Collection<FileStatus>) statuses);
   }
@@ -509,7 +536,9 @@ public List<Footer> getFooters(Configuration configuration, List<FileStatus> sta
   public List<Footer> getFooters(Configuration configuration, Collection<FileStatus> statuses) throws IOException {
     LOG.debug("reading {} files", statuses.size());
     boolean taskSideMetaData = isTaskSideMetaData(configuration);
-    return ParquetFileReader.readAllFootersInParallelUsingSummaryFiles(configuration, statuses, taskSideMetaData);
+    List<Footer> footers =
+        ParquetFileReader.readAllFootersInParallelUsingSummaryFiles(configuration, statuses, taskSideMetaData);
+    return withFileStatuses(footers, statuses);
   }
 
   /**
@@ -732,8 +761,6 @@ List<ParquetInputSplit> getSplits(
     for (Footer footer : footers) {
       final Path file = footer.getFile();
       LOG.debug("{}", file);
-      FileSystem fs = file.getFileSystem(configuration);
-      FileStatus fileStatus = fs.getFileStatus(file);
       ParquetMetadata parquetMetaData = footer.getParquetMetadata();
       List<BlockMetaData> blocks = parquetMetaData.getBlocks();
 
@@ -748,6 +775,11 @@ List<ParquetInputSplit> getSplits(
         continue;
       }
 
+      FileSystem fs = file.getFileSystem(configuration);
+      FileStatus fileStatus = ParquetInputFormat.getFileStatus(footer);
+      if (fileStatus == null) {
+        fileStatus = fs.getFileStatus(file);
+      }
       BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
       splits.addAll(generateSplits(
           filteredBlocks,
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatFileStatusReuse.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatFileStatusReuse.java
new file mode 100644
index 0000000000..84e40a386c
--- /dev/null
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestInputFormatFileStatusReuse.java
@@ -0,0 +1,295 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.parquet.hadoop;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RawLocalFileSystem;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetOutputFormat.JobSummaryLevel;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.example.GroupReadSupport;
+import org.apache.parquet.hadoop.example.GroupWriteSupport;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestInputFormatFileStatusReuse {
+  private static final String TRACKING_SCHEME = "tracking";
+  private static final MessageType SCHEMA = MessageTypeParser.parseMessageType("message test { required int32 id; }");
+
+  @Rule
+  public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Test
+  public void reusesListedFileStatusAfterReadingAndCachingFooters() throws Exception {
+    File parquetFile = writeParquetFile();
+    TestContext context = newTestContext(parquetFile);
+    FixedStatusInputFormat inputFormat = new FixedStatusInputFormat(context.status);
+
+    assertUsesListedStatus(inputFormat, context, context.status);
+    assertEquals(1, inputFormat.getFooterReadCount());
+
+    FileStatus refreshedStatus = new FileStatus(context.status);
+    inputFormat.setStatus(refreshedStatus);
+    context.fileSystem.clearObservations();
+    assertUsesListedStatus(inputFormat, context, refreshedStatus);
+    assertEquals("the second split plan should use the cached footer", 1, inputFormat.getFooterReadCount());
+  }
+
+  @Test
+  public void reusesListedFileStatusWithSummaryFileFooters() throws Exception {
+    File parquetFile = writeParquetFile();
+    writeSummaryFiles(parquetFile);
+    TestContext context = newTestContext(parquetFile);
+    FixedStatusInputFormat inputFormat = new FixedStatusInputFormat(context.status);
+
+    assertUsesListedStatus(inputFormat, context, context.status);
+
+    Path metadataPath = new Path(context.status.getPath().getParent(), ParquetFileWriter.PARQUET_METADATA_FILE);
+    assertTrue("the summary metadata should supply the footer", context.fileSystem.getOpenCount(metadataPath) > 0);
+    assertEquals(
+        "the data file should not be opened when its footer comes from summary metadata",
+        0,
+        context.fileSystem.getOpenCount(context.status.getPath()));
+  }
+
+  @Test
+  @SuppressWarnings("deprecation")
+  public void looksUpFileStatusForFootersWithoutListedStatuses() throws Exception {
+    File parquetFile = writeParquetFile();
+    TestContext context = newTestContext(parquetFile);
+    Path localPath = new Path(parquetFile.toURI());
+    ParquetMetadata metadata = ParquetFileReader.readFooter(new Configuration(), localPath);
+    Footer footer = new Footer(context.status.getPath(), metadata);
+    FixedStatusInputFormat inputFormat = new FixedStatusInputFormat(Collections.emptyList());
+
+    List<ParquetInputSplit> splits =
+        inputFormat.getSplits(context.job.getConfiguration(), Collections.singletonList(footer));
+
+    assertEquals(1, splits.size());
+    assertEquals(1, context.fileSystem.getFileStatusCount(context.status.getPath()));
+  }
+
+  @Test
+  @SuppressWarnings("deprecation")
+  public void reusesProvidedFileStatusWithDeprecatedSplitPlanning() throws Exception {
+    File parquetFile = writeParquetFile();
+    TestContext context = newTestContext(parquetFile);
+    FixedStatusInputFormat inputFormat = new FixedStatusInputFormat(context.status);
+    List<Footer> footers =
+        inputFormat.getFooters(context.job.getConfiguration(), Collections.singletonList(context.status));
+
+    context.fileSystem.clearObservations();
+    List<ParquetInputSplit> splits = inputFormat.getSplits(context.job.getConfiguration(), footers);
+
+    assertEquals(1, splits.size());
+    assertEquals(0, context.fileSystem.getFileStatusCount(context.status.getPath()));
+    assertSame(context.status, context.fileSystem.getBlockLocationStatus());
+  }
+
+  private void assertUsesListedStatus(
+      FixedStatusInputFormat inputFormat, TestContext context, FileStatus expectedStatus) throws IOException {
+    List<InputSplit> splits = inputFormat.getSplits(context.job);
+
+    assertEquals(1, splits.size());
+    assertEquals(
+        "split planning should not request status for a file returned by listStatus",
+        0,
+        context.fileSystem.getFileStatusCount(context.status.getPath()));
+    assertSame(
+        "block location lookup should use the FileStatus returned by listStatus",
+        expectedStatus,
+        context.fileSystem.getBlockLocationStatus());
+  }
+
+  private TestContext newTestContext(File parquetFile) throws Exception {
+    Configuration configuration = new Configuration();
+    configuration.setClass("fs." + TRACKING_SCHEME + ".impl", TrackingFileSystem.class, FileSystem.class);
+    Job job = Job.getInstance(configuration);
+    ParquetInputFormat.setTaskSideMetaData(job, false);
+
+    Path path = new Path(TRACKING_SCHEME, UUID.randomUUID().toString(), parquetFile.getAbsolutePath());
+    TrackingFileSystem fileSystem = (TrackingFileSystem) path.getFileSystem(configuration);
+    FileStatus status = fileSystem.getFileStatus(path);
+    fileSystem.clearObservations();
+    return new TestContext(job, fileSystem, status);
+  }
+
+  private File writeParquetFile() throws IOException {
+    File file = new File(temporaryFolder.getRoot(), "part-00000.parquet");
+    Configuration configuration = new Configuration();
+    GroupWriteSupport.setSchema(SCHEMA, configuration);
+
+    try (ParquetWriter<Group> writer = ExampleParquetWriter.builder(new Path(file.toURI()))
+        .withConf(configuration)
+        .withType(SCHEMA)
+        .build()) {
+      writer.write(new SimpleGroupFactory(SCHEMA).newGroup().append("id", 1));
+    }
+    return file;
+  }
+
+  private void writeSummaryFiles(File parquetFile) throws IOException {
+    Configuration configuration = new Configuration();
+    Path file = new Path(parquetFile.toURI());
+    FileStatus status = file.getFileSystem(configuration).getFileStatus(file);
+    ParquetMetadata metadata = ParquetFileReader.readFooter(configuration, status);
+    Footer footer = new Footer(file, metadata);
+    ParquetFileWriter.writeMetadataFile(
+        configuration,
+        new Path(parquetFile.getParentFile().toURI()),
+        Collections.singletonList(footer),
+        JobSummaryLevel.ALL);
+  }
+
+  private static final class TestContext {
+    private final Job job;
+    private final TrackingFileSystem fileSystem;
+    private final FileStatus status;
+
+    private TestContext(Job job, TrackingFileSystem fileSystem, FileStatus status) {
+      this.job = job;
+      this.fileSystem = fileSystem;
+      this.status = status;
+    }
+  }
+
+  private static final class FixedStatusInputFormat extends ParquetInputFormat<Group> {
+    private List<FileStatus> statuses;
+    private int footerReadCount;
+
+    private FixedStatusInputFormat(FileStatus status) {
+      this(Collections.singletonList(status));
+    }
+
+    private FixedStatusInputFormat(List<FileStatus> statuses) {
+      super(GroupReadSupport.class);
+      this.statuses = statuses;
+    }
+
+    @Override
+    protected List<FileStatus> listStatus(org.apache.hadoop.mapreduce.JobContext jobContext) {
+      return statuses;
+    }
+
+    @Override
+    public List<Footer> getFooters(Configuration configuration, Collection<FileStatus> statuses)
+        throws IOException {
+      footerReadCount++;
+      return super.getFooters(configuration, statuses);
+    }
+
+    private int getFooterReadCount() {
+      return footerReadCount;
+    }
+
+    private void setStatus(FileStatus status) {
+      statuses = Collections.singletonList(status);
+    }
+  }
+
+  public static final class TrackingFileSystem extends RawLocalFileSystem {
+    private final RawLocalFileSystem delegate = new RawLocalFileSystem();
+    private final List<Path> fileStatusPaths = Collections.synchronizedList(new ArrayList<Path>());
+    private final List<Path> openedPaths = Collections.synchronizedList(new ArrayList<Path>());
+    private URI uri = URI.create(TRACKING_SCHEME + ":///");
+    private volatile FileStatus blockLocationStatus;
+
+    @Override
+    public void initialize(URI name, Configuration configuration) throws IOException {
+      super.initialize(name, configuration);
+      String authority = name.getAuthority() == null ? "" : name.getAuthority();
+      uri = URI.create(name.getScheme() + "://" + authority + "/");
+      delegate.initialize(URI.create("file:///"), configuration);
+    }
+
+    @Override
+    public URI getUri() {
+      return uri == null ? URI.create(TRACKING_SCHEME + ":///") : uri;
+    }
+
+    @Override
+    public FSDataInputStream open(Path path, int bufferSize) throws IOException {
+      openedPaths.add(path);
+      return delegate.open(toLocalPath(path), bufferSize);
+    }
+
+    @Override
+    public FileStatus getFileStatus(Path path) throws IOException {
+      fileStatusPaths.add(path);
+      FileStatus status = new FileStatus(delegate.getFileStatus(toLocalPath(path)));
+      status.setPath(path);
+      return status;
+    }
+
+    @Override
+    public BlockLocation[] getFileBlockLocations(FileStatus status, long start, long length) throws IOException {
+      blockLocationStatus = status;
+      return super.getFileBlockLocations(status, start, length);
+    }
+
+    private Path toLocalPath(Path path) {
+      return new Path(path.toUri().getPath());
+    }
+
+    private int getFileStatusCount(Path path) {
+      synchronized (fileStatusPaths) {
+        return Collections.frequency(fileStatusPaths, path);
+      }
+    }
+
+    private int getOpenCount(Path path) {
+      synchronized (openedPaths) {
+        return Collections.frequency(openedPaths, path);
+      }
+    }
+
+    private FileStatus getBlockLocationStatus() {
+      return blockLocationStatus;
+    }
+
+    private void clearObservations() {
+      fileStatusPaths.clear();
+      openedPaths.clear();
+      blockLocationStatus = null;
+    }
+  }
+}