Bump sparkey to 3.6.1, add LoadMode parameter to sparkeySideInput

Sparkey 3.6.1 adds page cache prefetch and mlock support.
This exposes it as an opt-in LoadMode parameter on sparkeySideInput,
allowing users to prefetch sparkey files into memory on first access.
ShardedSparkeyReader overrides load() using LoadResult.combine().

Usage: sc.sparkeySideInput(path, loadMode = LoadMode.ALL)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: spkrka

build.sbt

@@ -115,7 +115,7 @@ val scalaCollectionCompatVersion = "2.14.0"
 val scalaMacrosVersion = "2.1.1"
 val scalatestVersion = "3.2.19"
 val shapelessVersion = "2.3.13"
-val sparkeyVersion = "3.5.1"
+val sparkeyVersion = "3.6.1"
 val tensorFlowVersion = "1.1.0"
 val tensorFlowMetadataVersion = "1.16.1"
 val testContainersVersion = "0.44.1"

scio-extra/src/main/scala/com/spotify/scio/extra/sparkey/instances/ShardedSparkeyReader.scala

@@ -18,7 +18,7 @@ package com.spotify.scio.extra.sparkey.instances
 
 import java.util
 
-import com.spotify.sparkey.{IndexHeader, LogHeader, SparkeyReader}
+import com.spotify.sparkey.{IndexHeader, LoadMode, LoadResult, LogHeader, SparkeyReader}
 
 import scala.util.hashing.MurmurHash3
 import scala.jdk.CollectionConverters._
@@ -81,6 +81,9 @@ class ShardedSparkeyReader(val sparkeys: Map[Short, SparkeyReader], val numShard
   override def iterator(): util.Iterator[SparkeyReader.Entry] =
     sparkeys.values.map(_.iterator.asScala).reduce(_ ++ _).asJava
 
+  override def load(mode: LoadMode, executor: java.util.concurrent.Executor): LoadResult =
+    LoadResult.combine(sparkeys.values.map(_.load(mode, executor)).toArray: _*)
+
   override def getLoadedBytes: Long = sparkeys.valuesIterator.map(_.getLoadedBytes).sum
 
   override def getTotalBytes: Long = sparkeys.valuesIterator.map(_.getTotalBytes).sum

scio-extra/src/main/scala/com/spotify/scio/extra/sparkey/package.scala

@@ -26,7 +26,7 @@ import com.spotify.scio.io.{ClosedTap, EmptyTap}
 import com.spotify.scio.testing.TestDataManager
 import com.spotify.scio.util.{Cache, RemoteFileUtil}
 import com.spotify.scio.values.{SCollection, SideInput}
-import com.spotify.sparkey.{CompressionType, SparkeyReader}
+import com.spotify.sparkey.{CompressionType, LoadMode, SparkeyReader}
 import org.apache.beam.sdk.transforms.{DoFn, View}
 import org.apache.beam.sdk.util.CoderUtils
 import org.apache.beam.sdk.values.PCollectionView
@@ -113,7 +113,11 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
   /** Enhanced version of [[ScioContext]] with Sparkey methods. */
   implicit class SparkeyScioContext(private val self: ScioContext) extends AnyVal {
 
-    private def sparkeySideInput[T](basePath: String, mapFn: SparkeyReader => T): SideInput[T] = {
+    private def sparkeySideInput[T](
+      basePath: String,
+      mapFn: SparkeyReader => T,
+      loadMode: LoadMode
+    ): SideInput[T] = {
       if (self.isTest) {
         val id = self.testId.get
         val view = TestDataManager
@@ -126,7 +130,7 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
         val view: PCollectionView[SparkeyUri] = self
           .parallelize(paths)
           .applyInternal(View.asSingleton())
-        new SparkeySideInput(view, mapFn)
+        new SparkeySideInput(view, mapFn, loadMode)
       }
     }
 
@@ -135,10 +139,18 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
      * [[com.spotify.scio.values.SCollection.withSideInputs SCollection.withSideInputs]]. If the
      * provided base path ends with "*", it will be treated as a sharded collection of Sparkey
      * files.
+     *
+     * @param loadMode
+     *   (optional) page cache prefetch mode. Use `LoadMode.ALL` to prefetch both index and log
+     *   files into memory on first access, reducing page faults for large files on slow storage.
+     *   Default is `LoadMode.NONE` (no prefetching).
      */
     @experimental
-    def sparkeySideInput(basePath: String): SideInput[SparkeyReader] =
-      sparkeySideInput(basePath, identity)
+    def sparkeySideInput(
+      basePath: String,
+      loadMode: LoadMode = LoadMode.NONE
+    ): SideInput[SparkeyReader] =
+      sparkeySideInput(basePath, identity, loadMode)
 
     /**
      * Create a SideInput of `TypedSparkeyReader` from a [[SparkeyUri]] base path, to be used with
@@ -159,7 +171,8 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
             reader,
             decoder,
             Option(cache).getOrElse(Cache.noOp[String, T])
-          )
+          ),
+        LoadMode.NONE
       )
 
     /**
@@ -171,7 +184,11 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
       basePath: String,
       cache: Cache[String, String]
     ): SideInput[CachedStringSparkeyReader] =
-      sparkeySideInput(basePath, reader => new CachedStringSparkeyReader(reader, cache))
+      sparkeySideInput(
+        basePath,
+        reader => new CachedStringSparkeyReader(reader, cache),
+        LoadMode.NONE
+      )
   }
 
   /** Enhanced version of [[com.spotify.scio.values.SCollection SCollection]] with Sparkey methods. */
@@ -476,7 +493,7 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
      */
     @experimental
     def asSparkeySideInput: SideInput[SparkeyReader] =
-      new SparkeySideInput(self.applyInternal(View.asSingleton()), identity)
+      new SparkeySideInput(self.applyInternal(View.asSingleton()), identity, LoadMode.NONE)
 
     /**
      * Convert this SCollection to a SideInput of `SparkeyReader`, to be used with
@@ -490,7 +507,8 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
     ): SideInput[TypedSparkeyReader[T]] = {
       new SparkeySideInput(
         self.applyInternal(View.asSingleton()),
-        reader => new TypedSparkeyReader[T](reader, decoder, cache)
+        reader => new TypedSparkeyReader[T](reader, decoder, cache),
+        LoadMode.NONE
       )
     }
 
@@ -504,7 +522,8 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
     def asTypedSparkeySideInput[T](decoder: Array[Byte] => T): SideInput[TypedSparkeyReader[T]] =
       new SparkeySideInput(
         self.applyInternal(View.asSingleton()),
-        reader => new TypedSparkeyReader[T](reader, decoder, Cache.noOp)
+        reader => new TypedSparkeyReader[T](reader, decoder, Cache.noOp),
+        LoadMode.NONE
       )
 
     /**
@@ -517,7 +536,8 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
     ): SideInput[CachedStringSparkeyReader] =
       new SparkeySideInput(
         self.applyInternal(View.asSingleton()),
-        reader => new CachedStringSparkeyReader(reader, cache)
+        reader => new CachedStringSparkeyReader(reader, cache),
+        LoadMode.NONE
       )
   }
 
@@ -542,15 +562,17 @@ package object sparkey extends SparkeyReaderInstances with SparkeyCoders {
 
   private class SparkeySideInput[T](
     val view: PCollectionView[SparkeyUri],
-    mapFn: SparkeyReader => T
+    mapFn: SparkeyReader => T,
+    loadMode: LoadMode
   ) extends SideInput[T] {
     override def updateCacheOnGlobalWindow: Boolean = false
-    override def get[I, O](context: DoFn[I, O]#ProcessContext): T =
-      mapFn(
-        SparkeySideInput.checkMemory(
-          context.sideInput(view).getReader(RemoteFileUtil.create(context.getPipelineOptions))
-        )
+    override def get[I, O](context: DoFn[I, O]#ProcessContext): T = {
+      val reader = SparkeySideInput.checkMemory(
+        context.sideInput(view).getReader(RemoteFileUtil.create(context.getPipelineOptions))
       )
+      reader.load(loadMode)
+      mapFn(reader)
+    }
   }
 
   /**