fix comments

Author: steFaiz

paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala

@@ -373,8 +373,12 @@ case class MergeIntoPaimonDataEvolutionTable(
     // The final output is composed by updated columns, metadata columns and blob marker columns.
     // Marker columns are used to mark whether a blob field should be written with placeholder
     val rawBlobUpdateColumns = updateColumnsSorted.filter(isRawBlobUpdateColumn)
-    val rawBlobMarkerNamesByColumn = rawBlobUpdateColumns.zipWithIndex.map {
-      case (attr, index) => attr.name -> rawBlobMarkerName(index)
+    val rawBlobMarkerNames =
+      rawBlobMarkerNamesAvoiding(
+        rawBlobUpdateColumns.size,
+        updateColumnsSorted.map(_.name) ++ sourceTable.output.map(_.name))
+    val rawBlobMarkerNamesByColumn = rawBlobUpdateColumns.zip(rawBlobMarkerNames).map {
+      case (attr, markerName) => attr.name -> markerName
     }.toMap
     val rawBlobMarkerAttributes = rawBlobUpdateColumns.map(
       attr =>
@@ -832,6 +836,22 @@ object MergeIntoPaimonDataEvolutionTable {
     RAW_BLOB_PLACEHOLDER_MARKER_PREFIX + index
   }
 
+  private[commands] def rawBlobMarkerNamesAvoiding(
+      count: Int,
+      reservedNames: Seq[String]): Seq[String] = {
+    var nextIndex = 0
+    (0 until count).map {
+      _ =>
+        var markerName = rawBlobMarkerName(nextIndex)
+        while (reservedNames.exists(reservedName => resolver(reservedName, markerName))) {
+          nextIndex += 1
+          markerName = rawBlobMarkerName(nextIndex)
+        }
+        nextIndex += 1
+        markerName
+    }
+  }
+
   private def quotedColumn(name: String) = {
     col("`" + name.replace("`", "``") + "`")
   }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala

@@ -372,8 +372,12 @@ case class MergeIntoPaimonDataEvolutionTable(
     // The final output is composed by updated columns, metadata columns and blob marker columns.
     // Marker columns are used to mark whether a blob field should be written with placeholder
     val rawBlobUpdateColumns = updateColumnsSorted.filter(isRawBlobUpdateColumn)
-    val rawBlobMarkerNamesByColumn = rawBlobUpdateColumns.zipWithIndex.map {
-      case (attr, index) => attr.name -> rawBlobMarkerName(index)
+    val rawBlobMarkerNames =
+      rawBlobMarkerNamesAvoiding(
+        rawBlobUpdateColumns.size,
+        updateColumnsSorted.map(_.name) ++ sourceTable.output.map(_.name))
+    val rawBlobMarkerNamesByColumn = rawBlobUpdateColumns.zip(rawBlobMarkerNames).map {
+      case (attr, markerName) => attr.name -> markerName
     }.toMap
     val rawBlobMarkerAttributes = rawBlobUpdateColumns.map(
       attr =>
@@ -831,6 +835,22 @@ object MergeIntoPaimonDataEvolutionTable {
     RAW_BLOB_PLACEHOLDER_MARKER_PREFIX + index
   }
 
+  private[commands] def rawBlobMarkerNamesAvoiding(
+      count: Int,
+      reservedNames: Seq[String]): Seq[String] = {
+    var nextIndex = 0
+    (0 until count).map {
+      _ =>
+        var markerName = rawBlobMarkerName(nextIndex)
+        while (reservedNames.exists(reservedName => resolver(reservedName, markerName))) {
+          nextIndex += 1
+          markerName = rawBlobMarkerName(nextIndex)
+        }
+        nextIndex += 1
+        markerName
+    }
+  }
+
   private def quotedColumn(name: String) = {
     col("`" + name.replace("`", "``") + "`")
   }

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/BlobUpdateTestBase.scala

@@ -99,6 +99,35 @@ class BlobUpdateTestBase extends PaimonSparkTestBase {
     }
   }
 
+  test("Blob: merge-into raw-data BLOB marker name does not collide with target column") {
+    withTable("s", "t") {
+      sql(
+        "CREATE TABLE t (id INT, `__paimon_raw_blob_placeholder_0` STRING, picture BINARY) " +
+          "TBLPROPERTIES ('row-tracking.enabled'='true', 'data-evolution.enabled'='true', " +
+          "'blob-field'='picture')")
+      sql("INSERT INTO t VALUES (1, 'old_marker_name', X'01'), (2, 'kept', X'02')")
+
+      sql("CREATE TABLE s (id INT, `__paimon_raw_blob_placeholder_0` STRING, picture BINARY)")
+      sql("INSERT INTO s VALUES (1, 'new_marker_name', X'4E4557')")
+
+      sql("""
+            |MERGE INTO t
+            |USING s
+            |ON t.id = s.id
+            |WHEN MATCHED THEN UPDATE SET
+            |  t.`__paimon_raw_blob_placeholder_0` = s.`__paimon_raw_blob_placeholder_0`,
+            |  t.picture = s.picture
+            |""".stripMargin)
+
+      checkAnswer(
+        sql("SELECT id, `__paimon_raw_blob_placeholder_0`, picture FROM t ORDER BY id"),
+        Seq(
+          Row(1, "new_marker_name", Array[Byte](78, 69, 87)),
+          Row(2, "kept", Array[Byte](2)))
+      )
+    }
+  }
+
   test("Blob: self merge updates raw-data BLOB column") {
     withTable("t") {
       sql("CREATE TABLE t (id INT, name STRING, picture BINARY) TBLPROPERTIES " +