Adapt SQL RAPIDS callers to columnar helpers

Signed-off-by: Gera Shegalov <gshegalov@nvidia.com>

Author: gerashegalov

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCoreDumpHandler.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2023, NVIDIA CORPORATION.
+ * Copyright (c) 2023-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,13 +29,32 @@ import org.apache.hadoop.fs.permission.{FsAction, FsPermission}
 
 import org.apache.spark.SparkContext
 import org.apache.spark.api.plugin.PluginContext
-import org.apache.spark.internal.Logging
 import org.apache.spark.io.CompressionCodec
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.rapids.execution.TrampolineUtil
 import org.apache.spark.util.SerializableConfiguration
 
-object GpuCoreDumpHandler extends Logging {
+object GpuCoreDumpHandler {
+  private val log = org.slf4j.LoggerFactory.getLogger(getClass.getName.stripSuffix("$"))
+
+  private def logInfo(msg: => String): Unit = {
+    if (log.isInfoEnabled) {
+      log.info(msg)
+    }
+  }
+
+  private def logWarning(msg: => String, throwable: Throwable): Unit = {
+    log.warn(msg, throwable)
+  }
+
+  private def logError(msg: => String): Unit = {
+    log.error(msg)
+  }
+
+  private def logError(msg: => String, throwable: Throwable): Unit = {
+    log.error(msg, throwable)
+  }
+
   private var executor: Option[ExecutorService] = None
   private var dumpedPath: Option[String] = None
   private var namedPipeFile: File = _

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuDeviceManager.scala

@@ -27,7 +27,6 @@ import com.nvidia.spark.rapids.jni.RmmSpark
 import com.nvidia.spark.rapids.spill.SpillFramework
 
 import org.apache.spark.{SparkConf, SparkEnv, TaskContext}
-import org.apache.spark.internal.Logging
 import org.apache.spark.network.util.ByteUnit
 import org.apache.spark.resource.ResourceInformation
 import org.apache.spark.sql.internal.SQLConf
@@ -38,7 +37,29 @@ private case object Initialized extends MemoryState
 private case object Uninitialized extends MemoryState
 private case object Errored extends MemoryState
 
-object GpuDeviceManager extends Logging {
+object GpuDeviceManager {
+  private val log = org.slf4j.LoggerFactory.getLogger(getClass.getName.stripSuffix("$"))
+
+  private def logDebug(msg: => String): Unit = {
+    if (log.isDebugEnabled) {
+      log.debug(msg)
+    }
+  }
+
+  private def logInfo(msg: => String): Unit = {
+    if (log.isInfoEnabled) {
+      log.info(msg)
+    }
+  }
+
+  private def logWarning(msg: => String): Unit = {
+    log.warn(msg)
+  }
+
+  private def logError(msg: => String): Unit = {
+    log.error(msg)
+  }
+
   // This config controls whether RMM/Pinned memory are initialized from the task
   // or from the executor side plugin. The default is to initialize from the
   // executor plugin.

sql-plugin/src/main/scala/com/nvidia/spark/rapids/InternalExclusiveModeGpuDiscoveryPlugin.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021, NVIDIA CORPORATION.
+ * Copyright (c) 2021-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,15 +24,30 @@ import ai.rapids.cudf.Cuda
 
 import org.apache.spark.SparkConf
 import org.apache.spark.api.resource.ResourceDiscoveryPlugin
-import org.apache.spark.internal.Logging
 import org.apache.spark.resource.{ResourceInformation, ResourceRequest}
 
 /**
  * Note, this class should not be referenced directly in source code.
  * It should be loaded by reflection using ShimLoader.newInstanceOf, see ./docs/dev/shims.md
  */
 protected class InternalExclusiveModeGpuDiscoveryPlugin
-  extends ResourceDiscoveryPlugin with Logging {
+  extends ResourceDiscoveryPlugin {
+
+  private val log = org.slf4j.LoggerFactory.getLogger(
+    classOf[InternalExclusiveModeGpuDiscoveryPlugin])
+
+  private def logInfo(msg: => String): Unit = {
+    if (log.isInfoEnabled) {
+      log.info(msg)
+    }
+  }
+
+  private def logWarning(msg: => String): Unit = {
+    if (log.isWarnEnabled) {
+      log.warn(msg)
+    }
+  }
+
   override def discoverResource(
     request: ResourceRequest,
     sparkconf: SparkConf

sql-plugin/src/main/scala/com/nvidia/spark/rapids/python/GpuPythonArguments.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2025, NVIDIA CORPORATION.
+ * Copyright (c) 2025-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,11 +27,11 @@ import org.apache.spark.sql.types.DataType
  * @param argOffsets The offsets of the original arguments in "flattenedArgs"
  * @param argNames The optional argument names
  */
-case class GpuPythonArguments(
-    flattenedArgs: Seq[Expression],
-    flattenedTypes: Seq[DataType],
-    argOffsets: Array[Array[Int]],
-    argNames: Option[Array[Array[Option[String]]]])
+class GpuPythonArguments(
+    val flattenedArgs: Seq[Expression],
+    val flattenedTypes: Seq[DataType],
+    val argOffsets: Array[Array[Int]],
+    val argNames: Option[Array[Array[Option[String]]]])
 
 /** Gpu version of ArgumentMetadata */
-case class GpuArgumentMeta(offset: Int, name: Option[String])
+class GpuArgumentMeta(val offset: Int, val name: Option[String])

sql-plugin/src/main/scala/com/nvidia/spark/rapids/python/PythonWorkerSemaphore.scala

@@ -26,7 +26,6 @@ import com.nvidia.spark.rapids.python.PythonConfEntries.CONCURRENT_PYTHON_WORKER
 import org.apache.commons.lang3.mutable.MutableInt
 
 import org.apache.spark.{SparkEnv, TaskContext}
-import org.apache.spark.internal.Logging
 
 /*
  * PythonWorkerSemaphore is used to limit the number of Python workers(processes) to be started
@@ -41,7 +40,15 @@ import org.apache.spark.internal.Logging
  * the inner semaphore when no longer needed.
  *
  */
-object PythonWorkerSemaphore extends Logging {
+object PythonWorkerSemaphore {
+  private val log = org.slf4j.LoggerFactory.getLogger(
+    "com.nvidia.spark.rapids.python.PythonWorkerSemaphore")
+
+  private def logDebug(msg: => String): Unit = {
+    if (log.isDebugEnabled) {
+      log.debug(msg)
+    }
+  }
 
   private lazy val rapidsConf = new RapidsConf(SparkEnv.get.conf)
   private lazy val workersPerGpu = rapidsConf.get(CONCURRENT_PYTHON_WORKERS)
@@ -97,7 +104,15 @@ object PythonWorkerSemaphore extends Logging {
   }
 }
 
-private final class PythonWorkerSemaphore(tasksPerGpu: Int) extends Logging {
+private final class PythonWorkerSemaphore(tasksPerGpu: Int) {
+  private val log = org.slf4j.LoggerFactory.getLogger(classOf[PythonWorkerSemaphore])
+
+  private def logDebug(msg: => String): Unit = {
+    if (log.isDebugEnabled) {
+      log.debug(msg)
+    }
+  }
+
   private val semaphore = new Semaphore(tasksPerGpu)
   // Map to track which tasks have acquired the semaphore.
   private val activeTasks = new ConcurrentHashMap[Long, MutableInt]

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/CoalesceConvertIterator.scala

@@ -22,7 +22,6 @@ import com.nvidia.spark.rapids.RapidsPluginImplicits._
 import com.nvidia.spark.rapids.hybrid.{CoalesceBatchConverter => NativeConverter, HybridHostRetryAllocator, RapidsHostColumn}
 
 import org.apache.spark.TaskContext
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.types.StructType
 import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnVector}
@@ -36,7 +35,13 @@ class CoalesceConvertIterator(cpuScanIter: Iterator[ColumnarBatch],
                               targetBatchSizeInBytes: Long,
                               schema: StructType,
                               metrics: Map[String, GpuMetric])
-  extends Iterator[Array[RapidsHostColumn]] with Logging {
+  extends Iterator[Array[RapidsHostColumn]] {
+
+  @transient private lazy val log = org.slf4j.LoggerFactory.getLogger(
+    classOf[CoalesceConvertIterator])
+
+  private def logInfo(msg: => String): Unit = if (log.isInfoEnabled) log.info(msg)
+
 
   private var converterImpl: NativeConverter = _
 
@@ -140,7 +145,7 @@ class CoalesceConvertIterator(cpuScanIter: Iterator[ColumnarBatch],
   }
 }
 
-object CoalesceConvertIterator extends Logging {
+object CoalesceConvertIterator {
   /**
    * Consumes the RapidsHostBatchProducer and converts the HostColumnVectors to Device ones.
    */

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/HybridExecutionUtils.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024-2025, NVIDIA CORPORATION.
+ * Copyright (c) 2024-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,7 +21,6 @@ import java.util.Locale
 import ai.rapids.cudf.DType
 import com.nvidia.spark.rapids.{RapidsConf, VersionUtils}
 
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, UnresolvedHint}
 import org.apache.spark.sql.catalyst.trees.TreePattern
@@ -33,7 +32,6 @@ import org.apache.spark.sql.rapids.execution.TrampolineUtil
 import org.apache.spark.sql.types._
 
 object HybridExecutionUtils extends PredicateHelper {
-  
   private val HYBRID_JAR_PLUGIN_CLASS_NAME = "com.nvidia.spark.rapids.hybrid.HybridPluginWrapper"
 
   /**
@@ -434,7 +432,7 @@ object HybridExecutionUtils extends PredicateHelper {
   }
 }
 
-object HybridExecOverrides extends Logging {
+object HybridExecOverrides {
   // The SQL hint enables HybridScan for specific tables even if HYBRID_PARQUET_READER is disabled
   val HYBRID_SCAN_HINT = "HYBRID_SCAN"
 

sql-plugin/src/main/scala/org/apache/spark/rapids/hybrid/RapidsHostBatchProducer.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2025, NVIDIA CORPORATION.
+ * Copyright (c) 2025-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,7 +25,6 @@ import com.nvidia.spark.rapids.hybrid.RapidsHostColumn
 import com.nvidia.spark.rapids.jni.RmmSpark
 
 import org.apache.spark.TaskContext
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.rapids.execution.TrampolineUtil
 
 /**
@@ -92,7 +91,15 @@ class PrefetchHostBatchProducer(
     taskAttId: Long,
     base: Iterator[Array[RapidsHostColumn]],
     capacity: Int,
-    waitTimeMetric: GpuMetric) extends RapidsHostBatchProducer with Logging {
+    waitTimeMetric: GpuMetric) extends RapidsHostBatchProducer {
+
+  @transient private lazy val log = org.slf4j.LoggerFactory.getLogger(
+    classOf[PrefetchHostBatchProducer])
+
+  private def logInfo(msg: => String): Unit = if (log.isInfoEnabled) log.info(msg)
+
+  private def logError(msg: => String): Unit = if (log.isErrorEnabled) log.error(msg)
+
 
   @volatile private var isInit: Boolean = false
   // Mark if there is in-progress element being produced in producerThread

sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuJsonScan.scala

@@ -268,15 +268,15 @@ case class GpuJsonScan(
     val broadcastedConf = sparkSession.sparkContext.broadcast(
       new SerializableConfiguration(hadoopConf))
 
-    GpuJsonPartitionReaderFactory(sparkSession.sessionState.conf, broadcastedConf,
+    new GpuJsonPartitionReaderFactory(sparkSession.sessionState.conf, broadcastedConf,
       dataSchema, readDataSchema, readPartitionSchema, parsedOptions, maxReaderBatchSizeRows,
       maxReaderBatchSizeBytes, maxGpuColumnSizeBytes, metrics, options.asScala.toMap)
   }
 
   override def withInputFile(): GpuScan = this
 }
 
-case class GpuJsonPartitionReaderFactory(
+class GpuJsonPartitionReaderFactory(
     sqlConf: SQLConf,
     broadcastedConf: Broadcast[SerializableConfiguration],
     dataSchema: StructType,
@@ -288,7 +288,8 @@ case class GpuJsonPartitionReaderFactory(
     maxReaderBatchSizeBytes: Long,
     maxGpuColumnSizeBytes: Long,
     metrics: Map[String, GpuMetric],
-    @transient params: Map[String, String]) extends ShimFilePartitionReaderFactory(params) {
+    @transient params: Map[String, String])
+  extends ShimFilePartitionReaderFactory(params) with Serializable {
 
   override def buildReader(partitionedFile: PartitionedFile): PartitionReader[InternalRow] = {
     throw new IllegalStateException("ROW BASED PARSING IS NOT SUPPORTED ON THE GPU...")

sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuReadJsonFileFormat.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2025, NVIDIA CORPORATION.
+ * Copyright (c) 2022-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -56,7 +56,7 @@ class GpuReadJsonFileFormat extends JsonFileFormat with GpuReadFileFormatWithMet
       sparkSession.sessionState.conf.sessionLocalTimeZone,
       sparkSession.sessionState.conf.columnNameOfCorruptRecord)
     val rapidsConf = new RapidsConf(sqlConf)
-    val factory = GpuJsonPartitionReaderFactory(
+    val factory = new GpuJsonPartitionReaderFactory(
       sqlConf,
       broadcastedHadoopConf,
       dataSchema,
@@ -81,7 +81,7 @@ class GpuReadJsonFileFormat extends JsonFileFormat with GpuReadFileFormatWithMet
   }
 }
 
-object GpuReadJsonFileFormat {
+object GpuReadJsonFileFormat extends Serializable {
   def tagSupport(meta: SparkPlanMeta[FileSourceScanExec]): Unit = {
     val fsse = meta.wrapped
     GpuJsonScan.tagSupport(