[Spark] Make MERGE source materialize depend on checkpointRDDBlockIdNotFoundError error class, not message

spark/src/main/scala-spark-3.5/shims/MergeIntoMaterializeSourceShims.scala

@@ -0,0 +1,27 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.commands.merge
+
+import org.apache.spark.SparkException
+
+object MergeIntoMaterializeSourceShims {
+
+  /** In Spark 3.5 we can only check for the error message :( */
+  def mergeMaterializedSourceRddBlockLostError(e: SparkException, rddId: Int): Boolean = {
+    e.getMessage.matches(s"(?s).*Checkpoint block rdd_${rddId}_[0-9]+ not found!.*")
+  }
+}

spark/src/main/scala-spark-master/shims/MergeIntoMaterializeSourceShims.scala

@@ -0,0 +1,28 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.delta.commands.merge
+
+import org.apache.spark.SparkException
+
+object MergeIntoMaterializeSourceShims {
+
+  /** In Spark 4.0+ we could check on error class, which is more stable. */
+  def mergeMaterializedSourceRddBlockLostError(e: SparkException, rddId: Int): Boolean = {
+    e.getErrorClass == "CHECKPOINT_RDD_BLOCK_ID_NOT_FOUND" &&
+      e.getMessageParameters.get("rddBlockId").contains(s"rdd_${rddId}")
+  }
+}

spark/src/main/scala/org/apache/spark/sql/delta/commands/merge/MergeIntoMaterializeSource.scala

@@ -176,8 +176,9 @@ trait MergeIntoMaterializeSource extends DeltaLogging with DeltaSparkPlanUtils {
     // SparkCoreErrors.checkpointRDDBlockIdNotFoundError from LocalCheckpointRDD.compute.
     case s: SparkException
       if materializedSourceRDD.nonEmpty &&
-        s.getMessage.matches(
-          mergeMaterializedSourceRddBlockLostErrorRegex(materializedSourceRDD.get.id)) =>
+        MergeIntoMaterializeSourceShims.mergeMaterializedSourceRddBlockLostError(
+          s,
+          materializedSourceRDD.get.id) =>
       logWarning(log"Materialized ${MDC(DeltaLogKeys.OPERATION, operation)} source RDD block " +
         log"lost. ${MDC(DeltaLogKeys.OPERATION, operation)} needs to be restarted. " +
         log"This was attempt number ${MDC(DeltaLogKeys.ATTEMPT, attempt)}.")
@@ -436,10 +437,6 @@ object MergeIntoMaterializeSource {
       MergeIntoMaterializeSourceReason.MATERIALIZED_REASONS.contains(materializeReason))
   }
 
-  // This depends on SparkCoreErrors.checkpointRDDBlockIdNotFoundError msg
-  def mergeMaterializedSourceRddBlockLostErrorRegex(rddId: Int): String =
-    s"(?s).*Checkpoint block rdd_${rddId}_[0-9]+ not found!.*"
-
   /**
    * @return The columns of the source plan that are used in this MERGE
    */

spark/src/test/scala/org/apache/spark/sql/delta/MergeIntoMaterializeSourceSuite.scala

@@ -24,7 +24,7 @@ import scala.util.control.NonFatal
 import com.databricks.spark.util.{Log4jUsageLogger, MetricDefinitions, UsageRecord}
 import org.apache.spark.sql.delta.DeltaTestUtils._
 import org.apache.spark.sql.delta.commands.merge.{MergeIntoMaterializeSourceError, MergeIntoMaterializeSourceErrorType, MergeIntoMaterializeSourceReason, MergeStats}
-import org.apache.spark.sql.delta.commands.merge.MergeIntoMaterializeSource.mergeMaterializedSourceRddBlockLostErrorRegex
+import org.apache.spark.sql.delta.commands.merge.MergeIntoMaterializeSourceShims
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
 import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
@@ -176,8 +176,9 @@ trait MergeIntoMaterializeSourceErrorTests extends MergeIntoMaterializeSourceMix
     }
     assert(ex.isInstanceOf[SparkException], ex)
     assert(
-      ex.getMessage().matches(mergeMaterializedSourceRddBlockLostErrorRegex(rdd.id)),
-      s"RDD id ${rdd.id}: Message: ${ex.getMessage}")
+      MergeIntoMaterializeSourceShims.mergeMaterializedSourceRddBlockLostError(
+        ex.asInstanceOf[SparkException],
+        rdd.id))
   }
 
   for {