[Kernel] [Refactor] Update ParquetHandler::readParquetFiles interface to return a different data type (delta-io#4802)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [ ] Spark
- [ ] Standalone
- [ ] Flink
- [x] Kernel
- [ ] Other (fill in here)

## Description
Update ParquetHandler::readParquetFiles interface. Change the element
type of returned CloseableIterator from ColumnarBatch to FileReadResult.

<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->

## How was this patch tested?

<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->

## Does this PR introduce _any_ user-facing changes?
Yes.

<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

Author: mmmyr

kernel/examples/kernel-examples/src/main/java/io/delta/kernel/examples/MultiThreadedTableReader.java

@@ -244,7 +244,7 @@ private Runnable workConsumer(int workerId) {
                         engine.getParquetHandler().readParquetFiles(
                             singletonCloseableIterator(fileStatus),
                             physicalReadSchema,
-                            Optional.empty());
+                            Optional.empty()).map(res -> res.getData());
 
                     try (
                         CloseableIterator<FilteredColumnarBatch> dataIter =

kernel/examples/kernel-examples/src/main/java/io/delta/kernel/examples/SingleThreadedTableReader.java

@@ -115,7 +115,7 @@ private int readData(StructType readSchema, Scan scan, int maxRowCount) throws I
                             engine.getParquetHandler().readParquetFiles(
                                 singletonCloseableIterator(fileStatus),
                                 physicalReadSchema,
-                                Optional.empty());
+                                Optional.empty()).map(res -> res.getData());
                         try (
                             CloseableIterator<FilteredColumnarBatch> transformedData =
                                 Scan.transformPhysicalData(

kernel/kernel-api/src/main/java/io/delta/kernel/engine/FileReadResult.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright (2025) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.kernel.engine;
+
+import io.delta.kernel.data.ColumnarBatch;
+import java.util.Objects;
+
+/**
+ * The result of reading a batch of data in a file.
+ *
+ * <p>Encapsulates both the data read (as a {@link ColumnarBatch}) and the full path of the file
+ * from which the data was read.
+ */
+public class FileReadResult {
+
+  private final ColumnarBatch data;
+  private final String filePath;
+
+  /**
+   * Constructs a {@code FileReadResult} object with the given data and file path.
+   *
+   * @param data the columnar batch of data read from the file
+   * @param filePath the path of the file from which the data was read
+   */
+  public FileReadResult(ColumnarBatch data, String filePath) {
+    this.data = Objects.requireNonNull(data, "data must not be null");
+    this.filePath = Objects.requireNonNull(filePath, "filePath must not be null");
+  }
+
+  /** @return {@link ColumnarBatch} of data that was read from the file. */
+  public ColumnarBatch getData() {
+    return data;
+  }
+
+  /** @return the path of the file that this data was read from. */
+  public String getFilePath() {
+    return filePath;
+  }
+}

kernel/kernel-api/src/main/java/io/delta/kernel/engine/ParquetHandler.java

@@ -56,12 +56,12 @@ public interface ParquetHandler {
    * @param predicate Optional predicate which the Parquet reader can optionally use to prune rows
    *     that don't satisfy the predicate. Because pruning is optional and may be incomplete, caller
    *     is still responsible apply the predicate on the data returned by this method.
-   * @return an iterator of {@link ColumnarBatch}s containing the data in columnar format. It is the
-   *     responsibility of the caller to close the iterator. The data returned is in the same as the
-   *     order of files given in {@code scanFileIter}.
+   * @return an iterator of {@link FileReadResult}s containing the data in columnar format along
+   *     with metadata. It is the responsibility of the caller to close the iterator. The data
+   *     returned is in the same as the order of files given in {@code scanFileIter}.
    * @throws IOException if an I/O error occurs during the read.
    */
-  CloseableIterator<ColumnarBatch> readParquetFiles(
+  CloseableIterator<FileReadResult> readParquetFiles(
       CloseableIterator<FileStatus> fileIter,
       StructType physicalSchema,
       Optional<Predicate> predicate)

kernel/kernel-api/src/main/java/io/delta/kernel/internal/replay/ActionsIterator.java

@@ -27,6 +27,7 @@
 import io.delta.kernel.data.ColumnVector;
 import io.delta.kernel.data.ColumnarBatch;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.engine.FileReadResult;
 import io.delta.kernel.expressions.*;
 import io.delta.kernel.internal.checkpoints.SidecarFile;
 import io.delta.kernel.internal.fs.Path;
@@ -228,7 +229,8 @@ private CloseableIterator<ColumnarBatch> getActionsIterFromSinglePartOrV2Checkpo
                       .readParquetFiles(
                           singletonCloseableIterator(file),
                           finalReadSchema,
-                          checkpointPredicateIncludingSidecars),
+                          checkpointPredicateIncludingSidecars)
+                      .map(FileReadResult::getData),
               "Reading parquet log file `%s` with readSchema=%s and predicate=%s",
               file,
               finalReadSchema,
@@ -353,8 +355,8 @@ private CloseableIterator<ActionWrapper> getNextActionsIter() {
                     () ->
                         engine
                             .getParquetHandler()
-                            .readParquetFiles(
-                                checkpointFiles, deltaReadSchema, checkpointPredicate),
+                            .readParquetFiles(checkpointFiles, deltaReadSchema, checkpointPredicate)
+                            .map(FileReadResult::getData),
                     "Reading checkpoint sidecars [%s] with readSchema=%s and predicate=%s",
                     checkpointFiles,
                     deltaReadSchema,

kernel/kernel-api/src/test/scala/io/delta/kernel/internal/SnapshotManagerSuite.scala

@@ -21,6 +21,7 @@ import scala.collection.JavaConverters._
 import scala.reflect.ClassTag
 
 import io.delta.kernel.data.{ColumnarBatch, ColumnVector}
+import io.delta.kernel.engine.FileReadResult
 import io.delta.kernel.exceptions.{InvalidTableException, TableNotFoundException}
 import io.delta.kernel.expressions.Predicate
 import io.delta.kernel.internal.checkpoints.{CheckpointInstance, CheckpointMetaData, SidecarFile}
@@ -135,6 +136,7 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
     topLevelFileTypes.foreach { topLevelFileType =>
       val v2Checkpoints =
         v2CheckpointFileStatuses(v2CheckpointSpec, topLevelFileType)
+
       val checkpointFiles = v2Checkpoints.flatMap {
         case (topLevelCheckpointFile, sidecars) =>
           Seq(topLevelCheckpointFile) ++ sidecars
@@ -160,11 +162,15 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
       }.getOrElse((Seq.empty, Seq.empty))
 
       val compactions = compactedFileStatuses(compactionVersions)
-
+      val mockSidecarParquetHandler = if (expectedSidecars.nonEmpty) {
+        new MockSidecarParquetHandler(expectedSidecars, expectedV2Checkpoint.head.getPath)
+      } else {
+        new BaseMockParquetHandler {}
+      }
       val logSegment = snapshotManager.getLogSegmentForVersion(
         createMockFSListFromEngine(
           listFromProvider(deltas ++ compactions ++ checkpointFiles)("/"),
-          new MockSidecarParquetHandler(expectedSidecars),
+          mockSidecarParquetHandler,
           new MockSidecarJsonHandler(expectedSidecars)),
         versionToLoad)
 
@@ -775,31 +781,39 @@ class SnapshotManagerSuite extends AnyFunSuite with MockFileSystemClientUtils {
 }
 
 trait SidecarIteratorProvider extends VectorTestUtils {
-  def singletonSidecarIterator(sidecars: Seq[FileStatus])
-      : CloseableIterator[ColumnarBatch] = Utils.singletonCloseableIterator(
-    new ColumnarBatch {
-      override def getSchema: StructType = SidecarFile.READ_SCHEMA
-
-      override def getColumnVector(ordinal: Int): ColumnVector = {
-        ordinal match {
-          case 0 => stringVector(sidecars.map(_.getPath)) // path
-          case 1 => longVector(sidecars.map(_.getSize): _*) // size
-          case 2 =>
-            longVector(sidecars.map(_.getModificationTime): _*); // modification time
-        }
-      }
 
-      override def getSize: Int = sidecars.length
-    })
+  private def buildSidecarBatch(sidecars: Seq[FileStatus]): ColumnarBatch = new ColumnarBatch {
+    override def getSchema: StructType = SidecarFile.READ_SCHEMA
+
+    override def getColumnVector(ordinal: Int): ColumnVector = ordinal match {
+      case 0 => stringVector(sidecars.map(_.getPath)) // path
+      case 1 => longVector(sidecars.map(_.getSize): _*) // size
+      case 2 => longVector(sidecars.map(_.getModificationTime): _*) // modification time
+    }
+
+    override def getSize: Int = sidecars.length
+  }
+
+  def singletonSidecarParquetIterator(sidecars: Seq[FileStatus], v2CheckpointFileName: String)
+      : CloseableIterator[FileReadResult] = {
+    val batch = buildSidecarBatch(sidecars)
+    Utils.singletonCloseableIterator(new FileReadResult(batch, v2CheckpointFileName))
+  }
+
+  // TODO: [delta-io/delta#4849] extend FileReadResult for JSON read result
+  def singletonSidecarJsonIterator(sidecars: Seq[FileStatus]): CloseableIterator[ColumnarBatch] = {
+    val batch = buildSidecarBatch(sidecars)
+    Utils.singletonCloseableIterator(batch)
+  }
 }
 
-class MockSidecarParquetHandler(sidecars: Seq[FileStatus])
+class MockSidecarParquetHandler(sidecars: Seq[FileStatus], v2CheckpointFileName: String)
     extends BaseMockParquetHandler with SidecarIteratorProvider {
   override def readParquetFiles(
       fileIter: CloseableIterator[FileStatus],
       physicalSchema: StructType,
-      predicate: Optional[Predicate]): CloseableIterator[ColumnarBatch] =
-    singletonSidecarIterator(sidecars)
+      predicate: Optional[Predicate]): CloseableIterator[FileReadResult] =
+    singletonSidecarParquetIterator(sidecars, v2CheckpointFileName)
 }
 
 class MockSidecarJsonHandler(sidecars: Seq[FileStatus])
@@ -809,7 +823,7 @@ class MockSidecarJsonHandler(sidecars: Seq[FileStatus])
       fileIter: CloseableIterator[FileStatus],
       physicalSchema: StructType,
       predicate: Optional[Predicate]): CloseableIterator[ColumnarBatch] =
-    singletonSidecarIterator(sidecars)
+    singletonSidecarJsonIterator(sidecars)
 }
 
 class MockReadLastCheckpointFileJsonHandler(

kernel/kernel-api/src/test/scala/io/delta/kernel/test/MockEngineUtils.scala

@@ -107,7 +107,7 @@ trait BaseMockParquetHandler extends ParquetHandler with MockEngineUtils {
   override def readParquetFiles(
       fileIter: CloseableIterator[FileStatus],
       physicalSchema: StructType,
-      predicate: Optional[Predicate]): CloseableIterator[ColumnarBatch] =
+      predicate: Optional[Predicate]): CloseableIterator[FileReadResult] =
     throw new UnsupportedOperationException("not supported in this test suite")
 
   override def writeParquetFiles(

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/engine/DefaultParquetHandler.java

@@ -20,6 +20,7 @@
 import io.delta.kernel.defaults.engine.fileio.FileIO;
 import io.delta.kernel.defaults.internal.parquet.ParquetFileReader;
 import io.delta.kernel.defaults.internal.parquet.ParquetFileWriter;
+import io.delta.kernel.engine.FileReadResult;
 import io.delta.kernel.engine.ParquetHandler;
 import io.delta.kernel.expressions.Column;
 import io.delta.kernel.expressions.Predicate;
@@ -46,14 +47,15 @@ public DefaultParquetHandler(FileIO fileIO) {
   }
 
   @Override
-  public CloseableIterator<ColumnarBatch> readParquetFiles(
+  public CloseableIterator<FileReadResult> readParquetFiles(
       CloseableIterator<FileStatus> fileIter,
       StructType physicalSchema,
       Optional<Predicate> predicate)
       throws IOException {
-    return new CloseableIterator<ColumnarBatch>() {
+    return new CloseableIterator<FileReadResult>() {
       private final ParquetFileReader batchReader = new ParquetFileReader(fileIO);
       private CloseableIterator<ColumnarBatch> currentFileReader;
+      private String currentFilePath;
 
       @Override
       public void close() throws IOException {
@@ -70,8 +72,11 @@ public boolean hasNext() {
           // read.
           Utils.closeCloseables(currentFileReader);
           currentFileReader = null;
+          currentFilePath = null;
           if (fileIter.hasNext()) {
-            currentFileReader = batchReader.read(fileIter.next(), physicalSchema, predicate);
+            FileStatus fileStatus = fileIter.next();
+            currentFileReader = batchReader.read(fileStatus, physicalSchema, predicate);
+            currentFilePath = fileStatus.getPath();
             return hasNext(); // recurse since it's possible the loaded file is empty
           } else {
             return false;
@@ -80,8 +85,8 @@ public boolean hasNext() {
       }
 
       @Override
-      public ColumnarBatch next() {
-        return currentFileReader.next();
+      public FileReadResult next() {
+        return new FileReadResult(currentFileReader.next(), currentFilePath);
       }
     };
   }

kernel/kernel-defaults/src/test/java/io/delta/kernel/defaults/benchmarks/BenchmarkParallelCheckpointReading.java

@@ -26,6 +26,7 @@
 import io.delta.kernel.defaults.engine.hadoopio.HadoopFileIO;
 import io.delta.kernel.defaults.internal.parquet.ParquetFileReader;
 import io.delta.kernel.engine.Engine;
+import io.delta.kernel.engine.FileReadResult;
 import io.delta.kernel.engine.ParquetHandler;
 import io.delta.kernel.expressions.Predicate;
 import io.delta.kernel.internal.util.Utils;
@@ -177,17 +178,17 @@ static class ParallelParquetHandler extends DefaultParquetHandler {
     }
 
     @Override
-    public CloseableIterator<ColumnarBatch> readParquetFiles(
+    public CloseableIterator<FileReadResult> readParquetFiles(
         CloseableIterator<FileStatus> fileIter,
         StructType physicalSchema,
         Optional<Predicate> predicate)
         throws IOException {
-      return new CloseableIterator<ColumnarBatch>() {
+      return new CloseableIterator<FileReadResult>() {
         // Executor service will be closed as part of the returned `CloseableIterator`'s
         // close method.
         private final ExecutorService executorService = newFixedThreadPool(numberOfParallelThreads);
-        private Iterator<Future<List<ColumnarBatch>>> futuresIter;
-        private Iterator<ColumnarBatch> currentBatchIter;
+        private Iterator<Future<List<FileReadResult>>> futuresIter;
+        private Iterator<FileReadResult> currentBatchIter;
 
         @Override
         public void close() throws IOException {
@@ -213,15 +214,15 @@ public boolean hasNext() {
         }
 
         @Override
-        public ColumnarBatch next() {
+        public FileReadResult next() {
           return currentBatchIter.next();
         }
 
         private void submitReadRequestsIfNotDone() {
           if (futuresIter != null) {
             return;
           }
-          List<Future<List<ColumnarBatch>>> futures = new ArrayList<>();
+          List<Future<List<FileReadResult>>> futures = new ArrayList<>();
           while (fileIter.hasNext()) {
             futures.add(
                 executorService.submit(() -> parquetFileReader(fileIter.next(), physicalSchema)));
@@ -231,13 +232,13 @@ private void submitReadRequestsIfNotDone() {
       };
     }
 
-    List<ColumnarBatch> parquetFileReader(FileStatus fileStatus, StructType readSchema) {
+    List<FileReadResult> parquetFileReader(FileStatus fileStatus, StructType readSchema) {
       ParquetFileReader reader = new ParquetFileReader(fileIO);
       try (CloseableIterator<ColumnarBatch> batchIter =
           reader.read(fileStatus, readSchema, Optional.empty())) {
-        List<ColumnarBatch> batches = new ArrayList<>();
+        List<FileReadResult> batches = new ArrayList<>();
         while (batchIter.hasNext()) {
-          batches.add(batchIter.next());
+          batches.add(new FileReadResult(batchIter.next(), fileStatus.getPath()));
         }
         return batches;
       } catch (IOException e) {

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/LogReplayBaseSuite.scala

@@ -27,6 +27,7 @@ import io.delta.kernel.defaults.engine.fileio.FileIO
 import io.delta.kernel.defaults.engine.hadoopio.HadoopFileIO
 import io.delta.kernel.defaults.utils.TestUtils
 import io.delta.kernel.engine.{Engine, ExpressionHandler, FileSystemClient}
+import io.delta.kernel.engine.FileReadResult
 import io.delta.kernel.expressions.Predicate
 import io.delta.kernel.internal.checkpoints.Checkpointer
 import io.delta.kernel.internal.fs.Path
@@ -258,7 +259,7 @@ class MetricsParquetHandler(fileIO: FileIO)
   override def readParquetFiles(
       fileIter: CloseableIterator[FileStatus],
       physicalSchema: StructType,
-      predicate: Optional[Predicate]): CloseableIterator[ColumnarBatch] = {
+      predicate: Optional[Predicate]): CloseableIterator[FileReadResult] = {
     val fileReadSet = fileIter.toSeq
     checkpointReadRequestSizes += fileReadSet.size
     super.readParquetFiles(