add storage flag for cached local relations

Author: pranavdev022

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -5139,6 +5139,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(false)
 
+  val ARTIFACT_MANAGER_CACHE_STORAGE_LEVEL =
+    buildConf("spark.sql.artifact.cacheStorageLevel")
+      .internal()
+      .doc("Storage level for cached blocks in artifact manager. Valid values are any " +
+        "StorageLevel name (e.g., MEMORY_AND_DISK_SER, DISK_ONLY, MEMORY_ONLY, etc.).")
+      .version("4.2.0")
+      .stringConf
+      .createWithDefault("MEMORY_AND_DISK_SER")
+
   val FAST_HASH_AGGREGATE_MAX_ROWS_CAPACITY_BIT =
     buildConf("spark.sql.codegen.aggregate.fastHashMap.capacityBit")
       .internal()

sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala

@@ -194,6 +194,17 @@ class ArtifactManager(session: SparkSession) extends AutoCloseable with Logging
         // (e.g., after clone), we should replace it.
         val existingBlock = hashToCachedIdMap.get(hash)
         if (existingBlock == null || existingBlock.id != blockId) {
+          val storageLevelStr = session.conf.get(
+            SQLConf.ARTIFACT_MANAGER_CACHE_STORAGE_LEVEL)
+          val storageLevel = try {
+            StorageLevel.fromString(storageLevelStr)
+          } catch {
+            case e: IllegalArgumentException =>
+              logWarning(
+                log"Invalid storage level '${MDC(LogKeys.STORAGE_LEVEL, storageLevelStr)}' " +
+                  log"for artifact cache, falling back to MEMORY_AND_DISK_SER", e)
+              StorageLevel.MEMORY_AND_DISK_SER
+          }
           val updater = blockManager.TempFileBasedBlockStoreUpdater(
             blockId = blockId,
             level = StorageLevel.MEMORY_AND_DISK_SER,