Enable features via config that are off by default in the profiler Au…

…toTuner (#668)

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/AutoTuner.scala

@@ -330,7 +330,8 @@ class RecommendationEntry(val name: String,
 class AutoTuner(
     val clusterProps: ClusterProperties,
     val appInfoProvider: AppSummaryInfoBaseProvider,
-    val platform: Platform)  extends Logging {
+    val platform: Platform,
+    unsupportedOperators: Seq[DriverLogUnsupportedOperators])  extends Logging {
 
   import AutoTuner._
 
@@ -350,11 +351,9 @@ class AutoTuner(
   }
 
   def getPropertyValue(key: String): Option[String] = {
-    val fromProfile = appInfoProvider.getProperty(key)
-    fromProfile match {
-      case None => Option(clusterProps.softwareProperties.get(key))
-      case Some(_) => fromProfile
-    }
+    val fromProfile = Option(appInfoProvider).flatMap(_.getProperty(key))
+    // If the value is not found above, fallback to cluster properties
+    fromProfile.orElse(Option(clusterProps.softwareProperties.get(key)))
   }
 
   def initRecommendations(): Unit = {
@@ -819,6 +818,20 @@ class AutoTuner(
     appendRecommendation("spark.sql.shuffle.partitions", s"$shufflePartitions")
   }
 
+  /**
+   * Analyzes unsupported driver logs and generates recommendations for configuration properties.
+   */
+  private def recommendFromDriverLogs(): Unit = {
+    // Iterate through unsupported operators' reasons and check for matching properties
+    unsupportedOperators.map(_.reason).foreach { operatorReason =>
+      recommendationsFromDriverLogs.collect {
+        case (config, recommendedValue) if operatorReason.contains(config) =>
+          appendRecommendation(config, recommendedValue)
+          appendComment(commentForExperimentalConfig(config))
+      }
+    }
+  }
+
   def appendOptionalComment(lookup: String, comment: String): Unit = {
     if (!skippedRecommendations.contains(lookup)) {
       appendComment(comment)
@@ -921,6 +934,9 @@ class AutoTuner(
         case (property, value) => appendRecommendation(property, value)
       }
     }
+    if (unsupportedOperators.nonEmpty) {
+      recommendFromDriverLogs()
+    }
     (toRecommendationsProfileResult, toCommentProfileResult)
   }
 }
@@ -970,6 +986,8 @@ object AutoTuner extends Logging {
   val DEF_READ_SIZE_THRESHOLD = 100 * 1024L * 1024L * 1024L
   val DEFAULT_WORKER_INFO_PATH = "./worker_info.yaml"
   val SUPPORTED_SIZE_UNITS: Seq[String] = Seq("b", "k", "m", "g", "t", "p")
+  private val DOC_URL: String = "https://nvidia.github.io/spark-rapids/docs/" +
+    "additional-functionality/advanced_configs.html#advanced-configuration"
 
   val commentsForMissingProps: Map[String, String] = Map(
     "spark.executor.memory" ->
@@ -1017,15 +1035,27 @@ object AutoTuner extends Logging {
         "  If the Spark RAPIDS jar is being bundled with your Spark\n" +
         "  distribution, this step is not needed.")
   )
+
+  // Recommended values for specific unsupported configurations
+  private val recommendationsFromDriverLogs: Map[String, String] = Map(
+    "spark.rapids.sql.incompatibleDateFormats.enabled" -> "true"
+  )
+
+  def commentForExperimentalConfig(config: String): String = {
+    s"Using $config does not guarantee to produce the same results as CPU. " +
+      s"Please refer to $DOC_URL."
+  }
+
   // the plugin jar is in the form of rapids-4-spark_scala_binary-(version)-*.jar
   val pluginJarRegEx: Regex = "rapids-4-spark_\\d\\.\\d+-(\\d{2}\\.\\d{2}\\.\\d+).*\\.jar".r
 
   private def handleException(
       ex: Exception,
       appInfo: AppSummaryInfoBaseProvider,
-      platform: Platform): AutoTuner = {
+      platform: Platform,
+      unsupportedOperators: Seq[DriverLogUnsupportedOperators]): AutoTuner = {
     logError("Exception: " + ex.getStackTrace.mkString("Array(", ", ", ")"))
-    val tuning = new AutoTuner(new ClusterProperties(), appInfo, platform)
+    val tuning = new AutoTuner(new ClusterProperties(), appInfo, platform, unsupportedOperators)
     val msg = ex match {
       case cEx: ConstructorException => cEx.getContext
       case _ => if (ex.getCause != null) ex.getCause.toString else ex.toString
@@ -1075,26 +1105,30 @@ object AutoTuner extends Logging {
   def buildAutoTunerFromProps(
       clusterProps: String,
       singleAppProvider: AppSummaryInfoBaseProvider,
-      platform: Platform = PlatformFactory.createInstance()): AutoTuner = {
+      platform: Platform = PlatformFactory.createInstance(),
+      unsupportedOperators: Seq[DriverLogUnsupportedOperators] = Seq.empty): AutoTuner = {
     try {
       val clusterPropsOpt = loadClusterPropertiesFromContent(clusterProps)
-      new AutoTuner(clusterPropsOpt.getOrElse(new ClusterProperties()), singleAppProvider, platform)
+      new AutoTuner(clusterPropsOpt.getOrElse(new ClusterProperties()), singleAppProvider, platform,
+        unsupportedOperators)
     } catch {
       case e: Exception =>
-        handleException(e, singleAppProvider, platform)
+        handleException(e, singleAppProvider, platform, unsupportedOperators)
     }
   }
 
   def buildAutoTuner(
       filePath: String,
       singleAppProvider: AppSummaryInfoBaseProvider,
-      platform: Platform = PlatformFactory.createInstance()): AutoTuner = {
+      platform: Platform = PlatformFactory.createInstance(),
+      unsupportedOperators: Seq[DriverLogUnsupportedOperators] = Seq.empty): AutoTuner = {
     try {
       val clusterPropsOpt = loadClusterProps(filePath)
-      new AutoTuner(clusterPropsOpt.getOrElse(new ClusterProperties()), singleAppProvider, platform)
+      new AutoTuner(clusterPropsOpt.getOrElse(new ClusterProperties()), singleAppProvider, platform,
+        unsupportedOperators)
     } catch {
       case e: Exception =>
-        handleException(e, singleAppProvider, platform)
+        handleException(e, singleAppProvider, platform, unsupportedOperators)
     }
   }
 

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileArgs.scala

@@ -15,7 +15,7 @@
  */
 package com.nvidia.spark.rapids.tool.profiling
 
-import com.nvidia.spark.rapids.tool.{PlatformFactory, PlatformNames}
+import com.nvidia.spark.rapids.tool.PlatformNames
 import org.rogach.scallop.{ScallopConf, ScallopOption}
 import org.rogach.scallop.exceptions.ScallopException
 

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/ProfileMain.scala

@@ -77,10 +77,10 @@ object ProfileMain extends Logging {
     }
 
     val profiler = new Profiler(hadoopConf, appArgs, enablePB)
-    profiler.profile(eventLogFsFiltered)
-    if (driverLog.nonEmpty){
-      profiler.profileDriver(driverLog)
+    if (driverLog.nonEmpty) {
+      profiler.profileDriver(driverLog, eventLogFsFiltered.isEmpty)
     }
+    profiler.profile(eventLogFsFiltered)
     (0, filteredLogs.size)
   }
 

core/src/main/scala/com/nvidia/spark/rapids/tool/profiling/Profiler.scala

@@ -124,15 +124,21 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
     progressBar.foreach(_.finishAll())
   }
 
-  def profileDriver(driverLogInfos: String): Unit = {
+  def profileDriver(driverLogInfos: String, eventLogsEmpty: Boolean): Unit = {
     val profileOutputWriter = new ProfileOutputWriter(s"$outputDir/driver",
       Profiler.DRIVER_LOG_NAME, numOutputRows, true)
-
     try {
       val driverLogProcessor = new DriverLogProcessor(driverLogInfos)
-      val unsupportedDrivers = driverLogProcessor.processDriverLog()
+      val unsupportedDriverOperators = driverLogProcessor.processDriverLog()
       profileOutputWriter.write(s"Unsupported operators in driver log",
-        unsupportedDrivers)
+        unsupportedDriverOperators)
+      if (eventLogsEmpty && useAutoTuner) {
+        // Since event logs are empty, AutoTuner will not run while processing event logs.
+        // We need to run it here explicitly.
+        val (properties, comments) = runAutoTuner(None, unsupportedDriverOperators)
+        profileOutputWriter.writeText("\n### A. Recommended Configuration ###\n")
+        profileOutputWriter.writeText(Profiler.getAutoTunerResultsAsString(properties, comments))
+      }
     } finally {
       profileOutputWriter.close()
     }
@@ -403,6 +409,26 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
       appLogPath, ioAnalysisMetrics), compareRes)
   }
 
+  /**
+   * A wrapper method to run the AutoTuner.
+   * @param appInfo     Summary of the application for tuning.
+   * @param unsupportedDriverOperators List of unsupported operators from driver log
+   */
+  private def runAutoTuner(appInfo: Option[ApplicationSummaryInfo],
+                           unsupportedDriverOperators: Seq[DriverLogUnsupportedOperators])
+  : (Seq[RecommendedPropertyResult], Seq[RecommendedCommentResult]) = {
+      val appInfoProvider = appInfo.map(new SingleAppSummaryInfoProvider(_)).orNull
+      val workerInfoPath = appArgs.workerInfo.getOrElse(AutoTuner.DEFAULT_WORKER_INFO_PATH)
+      val platform = appArgs.platform()
+      val autoTuner: AutoTuner = AutoTuner.buildAutoTuner(workerInfoPath, appInfoProvider,
+        PlatformFactory.createInstance(platform), unsupportedDriverOperators)
+
+      // The autotuner allows skipping some properties,
+      // e.g., getRecommendedProperties(Some(Seq("spark.executor.instances"))) skips the
+      // recommendation related to executor instances.
+      autoTuner.getRecommendedProperties()
+  }
+
   def writeOutput(profileOutputWriter: ProfileOutputWriter,
       appsSum: Seq[ApplicationSummaryInfo], outputCombined: Boolean,
       comparedRes: Option[CompareSummaryInfo] = None): Unit = {
@@ -464,7 +490,7 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
     } else {
       appsSum
     }
-    sums.foreach { app =>
+    sums.foreach { app: ApplicationSummaryInfo =>
       profileOutputWriter.writeText("### A. Information Collected ###")
       profileOutputWriter.write("Application Information", app.appInfo)
       profileOutputWriter.write("Application Log Path Mapping", app.appLogPath)
@@ -510,15 +536,7 @@ class Profiler(hadoopConf: Configuration, appArgs: ProfileArgs, enablePB: Boolea
         Some("Unsupported SQL Ops"))
 
       if (useAutoTuner) {
-        val workerInfoPath = appArgs.workerInfo.getOrElse(AutoTuner.DEFAULT_WORKER_INFO_PATH)
-        val platform = appArgs.platform()
-        val autoTuner: AutoTuner = AutoTuner.buildAutoTuner(workerInfoPath,
-          new SingleAppSummaryInfoProvider(app),
-          PlatformFactory.createInstance(platform))
-        // the autotuner allows skipping some properties
-        // e.g. getRecommendedProperties(Some(Seq("spark.executor.instances"))) skips the
-        // recommendation related to executor instances.
-        val (properties, comments) = autoTuner.getRecommendedProperties()
+        val (properties, comments) = runAutoTuner(Some(app), Seq.empty)
         profileOutputWriter.writeText("\n### D. Recommended Configuration ###\n")
         profileOutputWriter.writeText(Profiler.getAutoTunerResultsAsString(properties, comments))
       }

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualificationArgs.scala

@@ -15,7 +15,7 @@
  */
 package com.nvidia.spark.rapids.tool.qualification
 
-import com.nvidia.spark.rapids.tool.{PlatformFactory, PlatformNames}
+import com.nvidia.spark.rapids.tool.PlatformNames
 import org.rogach.scallop.{ScallopConf, ScallopOption}
 import org.rogach.scallop.exceptions.ScallopException
 

core/src/test/scala/com/nvidia/spark/rapids/tool/profiling/AutoTunerSuite.scala

@@ -1359,4 +1359,152 @@ class AutoTunerSuite extends FunSuite with BeforeAndAfterEach with Logging {
     // scalastyle:on line.size.limit
     assert(expectedResults == autoTunerOutput)
   }
+
+  test("Recommendations generated for unsupported operators from driver logs only") {
+    val customProps = mutable.LinkedHashMap(
+      "spark.executor.cores" -> "8",
+      "spark.executor.memory" -> "47222m",
+      "spark.rapids.sql.concurrentGpuTasks" -> "2",
+      "spark.task.resource.gpu.amount" -> "0.0625")
+    val unsupportedDriverOperators = Seq(
+      DriverLogUnsupportedOperators(
+        "FromUnixTime", 1,
+        "Only UTC zone id is supported. Actual default zone id: America/Los_Angeles; " +
+          "CORRECTED format 'yyyyMMdd' on the GPU is not guaranteed to produce the same " +
+          "results as Spark on CPU. Set spark.rapids.sql.incompatibleDateFormats.enabled=true " +
+          "to force onto GPU.")
+    )
+    val workerInfo = buildWorkerInfoAsString(Some(customProps))
+    val autoTuner: AutoTuner = AutoTuner.buildAutoTunerFromProps(workerInfo, null,
+      PlatformFactory.createInstance(), unsupportedDriverOperators)
+    val (properties, comments) = autoTuner.getRecommendedProperties()
+    val autoTunerOutput = Profiler.getAutoTunerResultsAsString(properties, comments)
+    // scalastyle:off line.size.limit
+    val expectedResults =
+      s"""|
+          |Spark Properties:
+          |--conf spark.rapids.sql.incompatibleDateFormats.enabled=true
+          |
+          |Comments:
+          |- 'spark.rapids.sql.incompatibleDateFormats.enabled' was not set.
+          |- AutoTuner recommendations only support eventlogs generated by Spark applications utilizing RAPIDS Accelerator for Apache Spark
+          |- RAPIDS Accelerator for Apache Spark jar is missing in "spark.plugins". Please refer to https://docs.nvidia.com/spark-rapids/user-guide/latest/getting-started/overview.html
+          |- ${AutoTuner.commentForExperimentalConfig("spark.rapids.sql.incompatibleDateFormats.enabled")}
+          |""".stripMargin
+    // scalastyle:on line.size.limit
+    assert(expectedResults == autoTunerOutput)
+  }
+
+  test("Recommendations generated for unsupported operators from driver and event logs") {
+    val customProps = mutable.LinkedHashMap(
+      "spark.executor.cores" -> "8",
+      "spark.executor.memory" -> "47222m",
+      "spark.rapids.sql.concurrentGpuTasks" -> "2",
+      "spark.task.resource.gpu.amount" -> "0.0625")
+    val unsupportedDriverOperators = Seq(
+      DriverLogUnsupportedOperators(
+        "FromUnixTime", 1,
+        "Only UTC zone id is supported. Actual default zone id: America/Los_Angeles; " +
+          "CORRECTED format 'yyyyMMdd' on the GPU is not guaranteed to produce the same " +
+          "results as Spark on CPU. Set spark.rapids.sql.incompatibleDateFormats.enabled=true " +
+          "to force onto GPU.")
+    )
+    val workerInfo = buildWorkerInfoAsString(Some(customProps))
+    val autoTuner: AutoTuner = AutoTuner.buildAutoTunerFromProps(workerInfo,
+      getGpuAppMockInfoProvider, PlatformFactory.createInstance(), unsupportedDriverOperators)
+    val (properties, comments) = autoTuner.getRecommendedProperties()
+    val autoTunerOutput = Profiler.getAutoTunerResultsAsString(properties, comments)
+    // scalastyle:off line.size.limit
+    val expectedResults =
+      s"""|
+          |Spark Properties:
+          |--conf spark.executor.cores=16
+          |--conf spark.executor.instances=8
+          |--conf spark.executor.memory=32768m
+          |--conf spark.executor.memoryOverhead=8396m
+          |--conf spark.rapids.memory.pinnedPool.size=4096m
+          |--conf spark.rapids.shuffle.multiThreaded.reader.threads=16
+          |--conf spark.rapids.shuffle.multiThreaded.writer.threads=16
+          |--conf spark.rapids.sql.incompatibleDateFormats.enabled=true
+          |--conf spark.rapids.sql.multiThreadedRead.numThreads=20
+          |--conf spark.shuffle.manager=com.nvidia.spark.rapids.spark311.RapidsShuffleManager
+          |--conf spark.sql.adaptive.advisoryPartitionSizeInBytes=128m
+          |--conf spark.sql.adaptive.coalescePartitions.minPartitionNum=128
+          |--conf spark.sql.files.maxPartitionBytes=512m
+          |--conf spark.sql.shuffle.partitions=200
+          |
+          |Comments:
+          |- 'spark.executor.instances' was not set.
+          |- 'spark.executor.memoryOverhead' was not set.
+          |- 'spark.rapids.memory.pinnedPool.size' was not set.
+          |- 'spark.rapids.shuffle.multiThreaded.reader.threads' was not set.
+          |- 'spark.rapids.shuffle.multiThreaded.writer.threads' was not set.
+          |- 'spark.rapids.sql.incompatibleDateFormats.enabled' was not set.
+          |- 'spark.rapids.sql.multiThreadedRead.numThreads' was not set.
+          |- 'spark.shuffle.manager' was not set.
+          |- 'spark.sql.adaptive.advisoryPartitionSizeInBytes' was not set.
+          |- 'spark.sql.adaptive.coalescePartitions.minPartitionNum' was not set.
+          |- 'spark.sql.adaptive.enabled' should be enabled for better performance.
+          |- 'spark.sql.files.maxPartitionBytes' was not set.
+          |- 'spark.sql.shuffle.partitions' was not set.
+          |- ${AutoTuner.classPathComments("rapids.jars.missing")}
+          |- ${AutoTuner.classPathComments("rapids.shuffle.jars")}
+          |- ${AutoTuner.commentForExperimentalConfig("spark.rapids.sql.incompatibleDateFormats.enabled")}
+          |""".stripMargin
+    // scalastyle:on line.size.limit
+    assert(expectedResults == autoTunerOutput)
+  }
+
+
+  test("Recommendations generated for empty unsupported operators from driver logs only") {
+    val customProps = mutable.LinkedHashMap(
+      "spark.executor.cores" -> "8",
+      "spark.executor.memory" -> "47222m",
+      "spark.rapids.sql.concurrentGpuTasks" -> "2",
+      "spark.task.resource.gpu.amount" -> "0.0625")
+    val workerInfo = buildWorkerInfoAsString(Some(customProps))
+    val autoTuner: AutoTuner = AutoTuner.buildAutoTunerFromProps(workerInfo, null,
+      PlatformFactory.createInstance(), Seq.empty)
+    val (properties, comments) = autoTuner.getRecommendedProperties()
+    val autoTunerOutput = Profiler.getAutoTunerResultsAsString(properties, comments)
+    // scalastyle:off line.size.limit
+    val expectedResults =
+      s"""|Cannot recommend properties. See Comments.
+          |
+          |Comments:
+          |- AutoTuner recommendations only support eventlogs generated by Spark applications utilizing RAPIDS Accelerator for Apache Spark
+          |- RAPIDS Accelerator for Apache Spark jar is missing in "spark.plugins". Please refer to https://docs.nvidia.com/spark-rapids/user-guide/latest/getting-started/overview.html
+          |""".stripMargin
+    // scalastyle:on line.size.limit
+    assert(expectedResults == autoTunerOutput)
+  }
+
+  test("Recommendations not generated for unsupported operators from driver logs") {
+    // This test does not generate any recommendations for the unsupported operator 'Literal'
+    val customProps = mutable.LinkedHashMap(
+      "spark.executor.cores" -> "8",
+      "spark.executor.memory" -> "47222m",
+      "spark.rapids.sql.concurrentGpuTasks" -> "2",
+      "spark.task.resource.gpu.amount" -> "0.0625")
+    val unsupportedDriverOperators = Seq(
+      DriverLogUnsupportedOperators(
+        "Literal", 3,
+        "expression Literal 1700518632630000 produces an unsupported type TimestampType")
+    )
+    val workerInfo = buildWorkerInfoAsString(Some(customProps))
+    val autoTuner: AutoTuner = AutoTuner.buildAutoTunerFromProps(workerInfo, null,
+      PlatformFactory.createInstance(), unsupportedDriverOperators)
+    val (properties, comments) = autoTuner.getRecommendedProperties()
+    val autoTunerOutput = Profiler.getAutoTunerResultsAsString(properties, comments)
+    // scalastyle:off line.size.limit
+    val expectedResults =
+      s"""|Cannot recommend properties. See Comments.
+          |
+          |Comments:
+          |- AutoTuner recommendations only support eventlogs generated by Spark applications utilizing RAPIDS Accelerator for Apache Spark
+          |- RAPIDS Accelerator for Apache Spark jar is missing in "spark.plugins". Please refer to https://docs.nvidia.com/spark-rapids/user-guide/latest/getting-started/overview.html
+          |""".stripMargin
+    // scalastyle:on line.size.limit
+    assert(expectedResults == autoTunerOutput)
+  }
 }