[SPARK-48458][SQL] Data loss in dynamic override fix

common/utils/src/main/resources/error/error-conditions.json

@@ -8892,6 +8892,11 @@
       "No enough memory for aggregation"
     ]
   },
+  "_LEGACY_ERROR_TEMP_3303" : {
+    "message" : [
+      "The command <command> is unsafe to run after spark session stopped"
+    ]
+  },
   "_LEGACY_ERROR_USER_RAISED_EXCEPTION" : {
     "message" : [
       "<errorMessage>"

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -2901,4 +2901,10 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       )
     )
   }
+
+  def unsafeCommandWhenSparkSessionStopped(command: String): SparkRuntimeException = {
+    new SparkRuntimeException(
+      errorClass = "_LEGACY_ERROR_TEMP_3303",
+      messageParameters = Map("command" -> command))
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala

@@ -112,12 +112,28 @@ case class InsertIntoHadoopFsRelationCommand(
     }
 
     val jobId = java.util.UUID.randomUUID().toString
+
+    val activeSession = Option(sparkSession).filter(_.isUsable)
     val committer = FileCommitProtocol.instantiate(
       sparkSession.sessionState.conf.fileCommitProtocolClass,
       jobId = jobId,
       outputPath = outputPath.toString,
       dynamicPartitionOverwrite = dynamicPartitionOverwrite)
 
+    def isSessionActiveAndUnchanged: Boolean = {
+      SparkSession.getActiveSession.zip(activeSession).exists {
+        case (sessionBefore, sessionAfter) => sessionBefore eq sessionAfter
+      }
+    }
+
+    // SPARK-48458: we need to ensure we have the same spark session that had been used
+    // during dynamicPartitionOverwrite initialization. Otherwise, there is a data loss risk
+    // if a fallback to the default static mode happens.
+    if (!isSessionActiveAndUnchanged) {
+      throw QueryExecutionErrors.unsafeCommandWhenSparkSessionStopped(command =
+        getClass.getSimpleName)
+    }
+
     val doInsertion = if (mode == SaveMode.Append) {
       true
     } else {