fix

Author: huan233usc

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

@@ -36,6 +36,7 @@
 import io.delta.kernel.utils.FileStatus;
 import io.delta.storage.LogStore;
 import java.io.*;
+import java.nio.channels.ClosedByInterruptException;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
 
@@ -113,6 +114,9 @@ public boolean hasNext() {
           }
           return nextLine != null;
         } catch (IOException ex) {
+          if (ex instanceof ClosedByInterruptException || ex instanceof InterruptedIOException) {
+            throw new UncheckedIOException(ex);
+          }
           throw new KernelEngineException(
               format("Error reading JSON file: %s", currentFile.getPath()), ex);
         }

kernel/kernel-defaults/src/main/java/io/delta/kernel/defaults/internal/parquet/ParquetFileReader.java

@@ -31,6 +31,9 @@
 import io.delta.kernel.utils.CloseableIterator;
 import io.delta.kernel.utils.FileStatus;
 import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.io.UncheckedIOException;
+import java.nio.channels.ClosedByInterruptException;
 import java.util.*;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.parquet.filter2.compat.FilterCompat;
@@ -91,6 +94,9 @@ public boolean hasNext() {
           hasNotConsumedNextElement = next != null;
           return hasNotConsumedNextElement;
         } catch (IOException ex) {
+          if (ex instanceof ClosedByInterruptException || ex instanceof InterruptedIOException) {
+            throw new UncheckedIOException(ex);
+          }
           throw new KernelEngineException(
               "Error reading Parquet file: " + fileStatus.getPath(), ex);
         }
@@ -158,6 +164,9 @@ protected ReadSupport<Object> getReadSupport() {
 
           } catch (IOException e) {
             Utils.closeCloseablesSilently(fileReader, reader);
+            if (e instanceof ClosedByInterruptException || e instanceof InterruptedIOException) {
+              throw new UncheckedIOException(e);
+            }
             throw new KernelEngineException(
                 "Error reading Parquet file: " + fileStatus.getPath(), e);
           }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/engine/DefaultJsonHandlerSuite.scala

@@ -15,7 +15,9 @@
  */
 package io.delta.kernel.defaults.engine
 
+import java.io.UncheckedIOException
 import java.math.{BigDecimal => JBigDecimal}
+import java.nio.channels.ClosedByInterruptException
 import java.nio.file.FileAlreadyExistsException
 import java.util.Optional
 
@@ -24,9 +26,12 @@ import scala.collection.JavaConverters._
 import io.delta.kernel.data.ColumnVector
 import io.delta.kernel.defaults.engine.hadoopio.HadoopFileIO
 import io.delta.kernel.defaults.utils.{DefaultVectorTestUtils, TestRow, TestUtils}
+import io.delta.kernel.exceptions.KernelEngineException
 import io.delta.kernel.internal.actions.CommitInfo
 import io.delta.kernel.internal.util.InternalUtils.singletonStringColumnVector
+import io.delta.kernel.internal.util.Utils.singletonCloseableIterator
 import io.delta.kernel.types._
+import io.delta.kernel.utils.FileStatus
 
 import org.apache.hadoop.conf.Configuration
 import org.scalatest.funsuite.AnyFunSuite
@@ -554,4 +559,22 @@ class DefaultJsonHandlerSuite extends AnyFunSuite with TestUtils with DefaultVec
     assert(commitInfo.getInCommitTimestamp === Optional.empty())
     assert(commitInfo.getTimestamp === 1740009523401L)
   }
+
+  test("ClosedByInterruptException propagates as UncheckedIOException, not KernelEngineException") {
+    val handler = new DefaultJsonHandler(throwingFileIO(new ClosedByInterruptException()))
+    val fileIter = singletonCloseableIterator(
+      FileStatus.of("file:///fake/_delta_log/00000000000000000000.json", 100, 0))
+    val iter = handler.readJsonFiles(
+      fileIter,
+      new StructType().add("value", StringType.STRING),
+      Optional.empty())
+
+    val ex = intercept[UncheckedIOException] {
+      iter.hasNext()
+    }
+    assert(
+      ex.getCause.isInstanceOf[ClosedByInterruptException],
+      s"Expected ClosedByInterruptException cause, got: ${ex.getCause.getClass}")
+    assert(!ex.isInstanceOf[KernelEngineException])
+  }
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/internal/parquet/ParquetFileReaderSuite.scala

@@ -15,14 +15,17 @@
  */
 package io.delta.kernel.defaults.internal.parquet
 
+import java.io.UncheckedIOException
 import java.math.BigDecimal
-import java.util.TimeZone
+import java.nio.channels.ClosedByInterruptException
+import java.util.{Optional, TimeZone}
 
 import io.delta.golden.GoldenTableUtils.{goldenTableFile, goldenTablePath}
 import io.delta.kernel.defaults.utils.{ExpressionTestUtils, TestRow}
+import io.delta.kernel.exceptions.KernelEngineException
 import io.delta.kernel.test.VectorTestUtils
 import io.delta.kernel.types._
-import io.delta.kernel.utils.MetadataColumnTestUtils
+import io.delta.kernel.utils.{FileStatus, MetadataColumnTestUtils}
 
 import org.apache.spark.sql.internal.SQLConf
 import org.scalatest.funsuite.AnyFunSuite
@@ -385,4 +388,21 @@ class ParquetFileReaderSuite extends AnyFunSuite
       readParquetFilesUsingKernel(parquetFilePath, readSchema), /* actual */
       readParquetFilesUsingSpark(parquetFilePath, readSchema) /* expected */ )
   }
+
+  test("ClosedByInterruptException propagates as UncheckedIOException, not KernelEngineException") {
+    val reader = new ParquetFileReader(throwingFileIO(new ClosedByInterruptException()))
+    val fileStatus = FileStatus.of("file:///fake/test.parquet", 100, 0)
+    val iter = reader.read(
+      fileStatus,
+      new StructType().add("value", StringType.STRING),
+      Optional.empty())
+
+    val ex = intercept[UncheckedIOException] {
+      iter.hasNext()
+    }
+    assert(
+      ex.getCause.isInstanceOf[ClosedByInterruptException],
+      s"Expected ClosedByInterruptException cause, got: ${ex.getCause.getClass}")
+    assert(!ex.isInstanceOf[KernelEngineException])
+  }
 }

kernel/kernel-defaults/src/test/scala/io/delta/kernel/defaults/utils/TestUtils.scala

@@ -28,6 +28,7 @@ import io.delta.golden.GoldenTableUtils
 import io.delta.kernel.{Scan, Snapshot, Table, TransactionCommitResult}
 import io.delta.kernel.data._
 import io.delta.kernel.defaults.engine.DefaultEngine
+import io.delta.kernel.defaults.engine.fileio.{FileIO, InputFile, OutputFile, SeekableInputStream}
 import io.delta.kernel.defaults.internal.data.vector.{DefaultGenericVector, DefaultStructVector}
 import io.delta.kernel.defaults.test.{AbstractTableManagerAdapter, LegacyTableManagerAdapter, TableManagerAdapter}
 import io.delta.kernel.engine.Engine
@@ -988,4 +989,26 @@ trait AbstractTestUtils
       case None => Optional.empty()
     }
   }
+
+  /**
+   * Creates a FileIO whose newInputFile().newStream() always throws the given exception.
+   * Useful for testing error handling in handlers.
+   */
+  def throwingFileIO(exception: => Exception): FileIO = {
+    new FileIO {
+      override def newInputFile(path: String, fileSize: Long): InputFile = new InputFile {
+        override def length(): Long = fileSize
+        override def path(): String = path
+        override def newStream(): SeekableInputStream = throw exception
+      }
+      override def listFrom(filePath: String): CloseableIterator[FileStatus] = null
+      override def getFileStatus(path: String): FileStatus = null
+      override def resolvePath(path: String): String = null
+      override def mkdirs(path: String): Boolean = false
+      override def newOutputFile(path: String): OutputFile = null
+      override def delete(path: String): Boolean = false
+      override def getConf(confKey: String): Optional[String] = Optional.empty()
+      override def copyFileAtomically(src: String, dst: String, overwrite: Boolean): Unit = {}
+    }
+  }
 }