fix: read local files without Hadoop in COPY INTO (JDK 23+) — R1FIX.4 / #183

core/build.sbt

@@ -66,3 +66,16 @@ json4s ++ mockito ++ avro ++ cloudConnectors ++ repl :+ "com.google.code.gson" %
 "com.typesafe.scala-logging" %% "scala-logging" % Versions.scalaLogging :+
 "io.delta" %% "delta-standalone" % Versions.delta :+
 "org.scalatest" %% "scalatest" % Versions.scalatest % Test
+
+// Issue #183: run the very same test suite on an arbitrary JDK without changing the compile JDK.
+//   sbt -Dtest.jdk.home=/Library/Java/JavaVirtualMachines/zulu-25.jdk/Contents/Home \
+//       "core/testOnly *LocalPathSpec *FileSourceSpec"
+// With the property unset this is a no-op: tests run in-process on the default JDK as before.
+Test / fork := sys.props.get("test.jdk.home").isDefined
+
+Test / javaHome := sys.props.get("test.jdk.home").map(file)
+
+// `Test / parallelExecution := false` at build.sbt:101 is a BARE top-level statement, i.e. scoped
+// to the ROOT project only — it is NOT inherited by `core`. Without this line `core`'s suites run
+// concurrently in one JVM and the issue-#183 guards become order-dependent (see LocalPathSpec).
+Test / parallelExecution := false

core/src/main/scala/app/softnetwork/elastic/client/file/LocalPath.scala

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2025 SOFTNETWORK
+ *
+ * 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 app.softnetwork.elastic.client.file
+
+import java.net.URI
+import java.nio.file.{Path, Paths}
+
+import scala.util.{Success, Try}
+import scala.util.matching.Regex
+
+/** Classifies a `COPY INTO … FROM '<path>'` source string as "on this machine's filesystem" or
+  * "somewhere Hadoop has to reach".
+  *
+  * Why this exists (issue #183): Hadoop's `FileSystem.get` reaches
+  * `UserGroupInformation.getCurrentUser()`, which calls
+  * `javax.security.auth.Subject.getSubject(AccessControlContext)`. JDK 23 re-specified that method
+  * to throw `UnsupportedOperationException` whenever a Security Manager is not allowed (the
+  * default), and JEP 486 (JDK 24) made it throw `("getSubject is not supported")` unconditionally
+  * while removing the `-Djava.security.manager=allow` escape hatch (the VM refuses to start with
+  * it). Verified against hadoop-common 3.4.2 bytecode — upgrading Hadoop does not help.
+  *
+  * Anything that only needs bytes off a local file therefore bypasses Hadoop entirely. Remote
+  * schemes (`s3a`, `s3`, `gs`, `abfs`, `abfss`, `wasb`, `wasbs`, `hdfs`, …) keep going through
+  * Hadoop unchanged, and so do Parquet and Delta reads, which genuinely need it.
+  *
+  * This object intentionally has NO Hadoop dependency so it can be unit-tested in isolation.
+  */
+object LocalPath {
+
+  /** A URI scheme is at least TWO characters before the colon.
+    *
+    * The lower bound is not cosmetic: `new URI("C:/data/x.jsonl")` parses `C` as a scheme, so a
+    * one-character prefix must never be treated as one — otherwise every Windows drive-letter path
+    * would be misrouted to Hadoop.
+    */
+  private val SchemePrefix: Regex = """^([A-Za-z][A-Za-z0-9+.-]+):""".r
+
+  private val FileScheme = "file"
+
+  /** Only these authorities denote "this machine". */
+  private val LocalAuthorities = Set("", "localhost")
+
+  /** Extracted so `LocalPathSpec` can exercise BOTH platforms deterministically on either OS. */
+  private[file] val onWindows: Boolean = java.io.File.separatorChar == '\\'
+
+  /** The URI scheme of `filePath`, lowercased, or `None` when it has none.
+    *
+    * This is the ONLY scheme parser in the file package. `HadoopConfigurationFactory.forPath` used
+    * to run its own (`Try(new URI(path).getScheme)`), which disagreed with this one in two ways
+    * that silently cost a user their credentials — see AD-10.
+    *
+    * A Windows drive letter is not a scheme: [[SchemePrefix]] requires at least two characters
+    * before the colon, so `C:/data/x.jsonl` yields `None`.
+    */
+  def scheme(filePath: String): Option[String] =
+    Option(filePath)
+      .filter(_.trim.nonEmpty)
+      .flatMap(raw => SchemePrefix.findFirstMatchIn(raw).map(_.group(1).toLowerCase))
+
+  /** Returns the local [[java.nio.file.Path]] denoted by `filePath`, or `None` when the path is not
+    * on this machine's filesystem and must be handled by Hadoop.
+    *
+    * Local means: no scheme at all (absolute, relative, or a Windows drive path), or the `file:`
+    * scheme with an empty or `localhost` authority.
+    *
+    * `~` is NOT expanded, and surrounding whitespace is NOT trimmed — both match Hadoop's
+    * `Path(String)` exactly. Pass an absolute path.
+    *
+    * Throws [[java.nio.file.InvalidPathException]] (a subclass of `IllegalArgumentException`) for a
+    * string this platform cannot represent as a path at all — see AD-8.
+    */
+  def resolve(filePath: String): Option[Path] =
+    // `filter(_.trim.nonEmpty)` rejects blank input WITHOUT rewriting the string: Hadoop preserves
+    // leading/trailing whitespace in a file name and so must we, or `COPY INTO … FROM '/tmp/x '`
+    // silently reads `/tmp/x`.
+    Option(filePath).filter(_.trim.nonEmpty).flatMap { raw =>
+      scheme(raw) match {
+        case None             => Some(Paths.get(raw))
+        case Some(FileScheme) => fromFileUri(raw) // `scheme` already lowercased it
+        case Some(_)          => None // remote scheme → Hadoop
+      }
+    }
+
+  private def fromFileUri(raw: String): Option[Path] = {
+    // `new URI` percent-decodes correctly (UTF-8, and unlike URLDecoder it does not turn '+' into a
+    // space) but rejects unencoded characters such as a literal space. When it rejects the input,
+    // the user did not percent-encode, so the remainder is already the literal path.
+    //
+    // A query or fragment disqualifies the URI reading. Hadoop's `Path(String)` takes "the rest of
+    // the string" as the path — "query & fragment not supported" — so `?` and `#` are ordinary file
+    // name characters to it (verified: `new Path("file:///a/report#1.jsonl").toUri.getPath` is
+    // `/a/report#1.jsonl`). Using `u.getPath` there would truncate to `/a/report` and silently read
+    // a DIFFERENT file. Fall through to the literal split, which keeps them.
+    val (authority, path) = Try(new URI(raw)) match {
+      case Success(u)
+          if u.getPath != null && u.getPath.nonEmpty &&
+            u.getQuery == null && u.getFragment == null =>
+        (Option(u.getAuthority).getOrElse(""), u.getPath)
+      case _ =>
+        // Also the branch for an OPAQUE `file:` URI (`file:relative/x.jsonl`), where `getPath` is
+        // null. Opaque URIs are taken literally — they are not percent-decoded.
+        splitLiteral(raw.substring(FileScheme.length + 1))
+    }
+
+    // Only an empty authority or `localhost` denotes this machine. Any other host keeps the
+    // pre-existing Hadoop behaviour rather than silently reinterpreting it as a local path.
+    if (!LocalAuthorities.contains(authority.toLowerCase)) None
+    else Option(path).filter(_.nonEmpty).map(p => Paths.get(stripDriveSlash(p, onWindows)))
+  }
+
+  /** Splits `//authority/path`, `///path` or `/path` (the part after `file:`) without URI parsing.
+    */
+  private def splitLiteral(rest: String): (String, String) =
+    if (rest.startsWith("//")) {
+      val afterSlashes = rest.substring(2)
+      afterSlashes.indexOf('/') match {
+        case -1  => (afterSlashes, "")
+        case idx => (afterSlashes.substring(0, idx), afterSlashes.substring(idx))
+      }
+    } else ("", rest)
+
+  /** `file:///C:/data/x.jsonl` yields the URI path `/C:/data/x.jsonl`; Windows needs the leading
+    * slash removed before `Paths.get` will accept it.
+    *
+    * Gated on the platform on purpose: on POSIX `/C:` is a perfectly legal directory name, and
+    * stripping the slash there would turn an absolute path into a CWD-relative one.
+    *
+    * `windows` is a parameter rather than a direct read of [[onWindows]] so both branches are
+    * unit-testable on either OS.
+    */
+  private[file] def stripDriveSlash(p: String, windows: Boolean): String =
+    if (
+      windows && p.length >= 3 && p.charAt(0) == '/' && p.charAt(2) == ':' &&
+      Character.isLetter(p.charAt(1))
+    ) p.substring(1)
+    else p
+}

core/src/main/scala/app/softnetwork/elastic/client/file/package.scala

@@ -32,10 +32,10 @@ import org.apache.parquet.hadoop.util.HadoopInputFile
 import io.delta.standalone.DeltaLog
 import io.delta.standalone.data.{CloseableIterator, RowRecord}
 import io.delta.standalone.types._
-import org.apache.parquet.io.SeekableInputStream
 import org.slf4j.{Logger, LoggerFactory}
 
-import java.io.{BufferedReader, InputStream, InputStreamReader}
+import java.io.{BufferedInputStream, BufferedReader, InputStream, InputStreamReader}
+import java.nio.file.{Files, Path => NioPath}
 import scala.concurrent.{blocking, ExecutionContext, Future}
 import scala.io.{Source => IoSource}
 import scala.util.{Failure, Success, Try}
@@ -56,6 +56,33 @@ package object file {
     conf
   }
 
+  /** Default read-ahead buffer, mirroring the `io.file.buffer.size` that [[hadoopConfiguration]]
+    * and `HadoopConfigurationFactory.base()` already set (64 KB).
+    */
+  private val DefaultBufferSize = 65536
+
+  /** Opens `filePath` for reading.
+    *
+    * Local paths (schemeless, or `file:` with an empty/`localhost` authority) are opened straight
+    * through `java.nio.file`, never touching Hadoop — see [[LocalPath]] and issue #183
+    * (`Subject.getSubject` throws on JDK 23+ by default and on JDK 24+ unconditionally, so Hadoop's
+    * `UserGroupInformation.getCurrentUser()` can no longer be called).
+    *
+    * Every other scheme keeps the pre-existing Hadoop path unchanged.
+    */
+  def openStream(filePath: String)(implicit conf: Configuration): InputStream =
+    LocalPath.resolve(filePath) match {
+      case Some(local) =>
+        // Hadoop's LocalFileSystem wraps its stream in a BufferedFSInputStream sized by
+        // `io.file.buffer.size`; `Files.newInputStream` is unbuffered, so without this every
+        // 8 KB BufferedReader/Jackson refill becomes a syscall. Preserve the tuned buffer.
+        new BufferedInputStream(
+          Files.newInputStream(local),
+          conf.getInt("io.file.buffer.size", DefaultBufferSize)
+        )
+      case None => HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+    }
+
   /** Base trait for file sources */
   sealed trait FileSource {
 
@@ -146,6 +173,46 @@ package object file {
     protected def validateFile(
       filePath: String,
       checkIsFile: Boolean = true
+    )(implicit conf: Configuration): Unit =
+      LocalPath.resolve(filePath) match {
+        case Some(local) => validateLocalPath(filePath, local, checkIsFile)
+        case None        => validateHadoopPath(filePath, checkIsFile)
+      }
+
+    /** Local fast path — never touches Hadoop (issue #183). Error messages and log lines are
+      * byte-for-byte identical to the Hadoop branch: `FileSourceSpec` asserts on them.
+      */
+    private def validateLocalPath(
+      filePath: String,
+      local: NioPath,
+      checkIsFile: Boolean
+    ): Unit = {
+      if (!Files.exists(local)) {
+        throw new IllegalArgumentException(s"File does not exist: $filePath")
+      }
+
+      if (checkIsFile && !Files.isRegularFile(local)) {
+        throw new IllegalArgumentException(s"Path is not a file: $filePath")
+      }
+
+      if (!checkIsFile && !Files.isDirectory(local)) {
+        throw new IllegalArgumentException(s"Path is not a directory: $filePath")
+      }
+
+      val length = if (checkIsFile) Files.size(local) else 0L
+
+      if (checkIsFile && length == 0) {
+        logger.warn(s"⚠️  File is empty: $filePath")
+      }
+
+      val pathType = if (checkIsFile) "file" else "directory"
+      val sizeInfo = if (checkIsFile) s"($length bytes)" else ""
+      logger.info(s"📁 Loading $pathType: $filePath $sizeInfo")
+    }
+
+    private def validateHadoopPath(
+      filePath: String,
+      checkIsFile: Boolean
     )(implicit conf: Configuration): Unit = {
       val path = new Path(filePath)
       val fs = FileSystem.get(path.toUri, conf)
@@ -335,7 +402,7 @@ package object file {
           create = () => {
             logger.info(s"📂 Opening JSON file: $filePath")
             Try {
-              val is: InputStream = HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+              val is: InputStream = openStream(filePath)
               new BufferedReader(new InputStreamReader(is, "UTF-8"))
             } match {
               case Success(reader) => reader
@@ -430,14 +497,13 @@ package object file {
 
       Source
         .unfoldResource[String, (InputStream, JsonParser)](
-          // Create: Open file via Hadoop and create JSON parser
+          // Create: Open file and create JSON parser
           create = () => {
-            logger.info(s"📂 Opening JSON Array file via Hadoop: $filePath")
+            logger.info(s"📂 Opening JSON Array file: $filePath")
             Try {
-              val is: SeekableInputStream =
-                HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+              val is: InputStream = openStream(filePath)
 
-              // Create Jackson parser on top of Hadoop SeekableInputStream
+              // Create Jackson parser on top of the input stream
               val parser = jsonFactory.createParser(is)
 
               // Expect array start
@@ -449,7 +515,7 @@ package object file {
                 )
               }
 
-              logger.info(s"📊 Started parsing JSON Array via Hadoop FS")
+              logger.info(s"📊 Started parsing JSON Array")
               (is, parser)
             } match {
               case Success(result) => result
@@ -499,7 +565,7 @@ package object file {
             }
           },
 
-          // Close: Close parser and Hadoop input stream
+          // Close: Close parser and input stream
           close = { case (inputStream, parser) =>
             Try {
               parser.close() // This also closes the underlying stream
@@ -510,12 +576,12 @@ package object file {
                 logger.warn(s"⚠️  Failed to close JSON Array parser: ${ex.getMessage}")
             }
 
-            // Ensure Hadoop stream is closed
+            // Ensure the stream is closed
             Try(inputStream.close()) match {
               case Success(_) =>
-                logger.debug(s"🔒 Closed Hadoop input stream for: $filePath")
+                logger.debug(s"🔒 Closed input stream for: $filePath")
               case Failure(ex) =>
-                logger.warn(s"⚠️  Failed to close Hadoop input stream: ${ex.getMessage}")
+                logger.warn(s"⚠️  Failed to close input stream: ${ex.getMessage}")
             }
           }
         )
@@ -544,7 +610,7 @@ package object file {
       Source
         .future(Future {
           blocking {
-            val is: InputStream = HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+            val is: InputStream = openStream(filePath)
             try {
               val arrayNode = mapper.readTree(is)
               if (!arrayNode.isArray) {
@@ -578,7 +644,7 @@ package object file {
           throw ex
       }
 
-      val is: InputStream = HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+      val is: InputStream = openStream(filePath)
 
       try {
         val arrayNode = mapper.readTree(is)
@@ -1106,9 +1172,14 @@ package object file {
       filePath: String
     )(implicit conf: Configuration = hadoopConfiguration): Boolean = {
       Try {
-        val fs = FileSystem.get(conf)
-        val deltaLogPath = new Path(filePath, "_delta_log")
-        fs.exists(deltaLogPath) && fs.getFileStatus(deltaLogPath).isDirectory
+        LocalPath.resolve(filePath) match {
+          case Some(local) =>
+            Files.isDirectory(local.resolve("_delta_log"))
+          case None =>
+            val fs = FileSystem.get(conf)
+            val deltaLogPath = new Path(filePath, "_delta_log")
+            fs.exists(deltaLogPath) && fs.getFileStatus(deltaLogPath).isDirectory
+        }
       }.getOrElse(false)
     }
 
@@ -1118,7 +1189,7 @@ package object file {
       filePath: String
     )(implicit conf: Configuration = hadoopConfiguration): FileFormat = {
       Try {
-        val is = HadoopInputFile.fromPath(new Path(filePath), conf).newStream()
+        val is = openStream(filePath)
         try {
           val reader = new BufferedReader(new InputStreamReader(is, "UTF-8"))
           val firstChar = reader.read().toChar
@@ -1173,13 +1244,15 @@ package object file {
 
     /** Returns a [[Configuration]] appropriate for the URI scheme embedded in `path`. */
     def forPath(path: String): Configuration = {
-      val scheme = Try(new java.net.URI(path).getScheme).getOrElse(null)
-      val conf = scheme match {
-        case "s3a" | "s3"                        => s3aConf()
-        case "abfs" | "abfss" | "wasb" | "wasbs" => azureConf()
-        case "gs"                                => gcsConf()
-        case "hdfs"                              => hdfsConf()
-        case _                                   => localConf()
+      // Scheme detection is LocalPath's (AD-10): it lowercases, and — unlike `new URI` — it does
+      // not throw on an unencoded space, so `S3A://b/x` and `s3a://b/my file.jsonl` both reach the
+      // S3 branch instead of silently falling through to localConf() with no credentials.
+      val conf = LocalPath.scheme(path) match {
+        case Some("s3a") | Some("s3")                                    => s3aConf()
+        case Some("abfs") | Some("abfss") | Some("wasb") | Some("wasbs") => azureConf()
+        case Some("gs")                                                  => gcsConf()
+        case Some("hdfs")                                                => hdfsConf()
+        case _                                                           => localConf()
       }
       loadUserXmlConf(conf)
       conf