[KYUUBI #2390] RuleEliminateMarker stays in analyze phase for data masking

### _Why are the changes needed?_

Avoid creating duplicate data masking projection to prevent errors in #2390

### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2394 from yaooqinn/2390.

Closes #2390

31c5d07 [Kent Yao] [KYUUBI #2390] RuleEliminateMarker stays in analyze phase for data masking
ad0138c [Kent Yao] [KYUUBI #2390] RuleEliminateMarker stays in analyze phase for data masking

Authored-by: Kent Yao <yao@apache.org>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: yaooqinn

extensions/spark/kyuubi-spark-authz/src/main/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtension.scala

@@ -40,7 +40,7 @@ class RangerSparkExtension extends (SparkSessionExtensions => Unit) {
 
   override def apply(v1: SparkSessionExtensions): Unit = {
     v1.injectResolutionRule(new RuleApplyRowFilterAndDataMasking(_))
-    v1.injectPostHocResolutionRule(_ => new RuleEliminateMarker())
+    v1.injectOptimizerRule(_ => new RuleEliminateMarker())
     v1.injectOptimizerRule(new RuleAuthorization(_))
   }
 }

extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/RangerSparkExtensionSuite.scala

@@ -235,6 +235,26 @@ abstract class RangerSparkExtensionSuite extends KyuubiFunSuite with SparkSessio
       doAs("admin", sql(s"DROP TABLE IF EXISTS $db.$table"))
     }
   }
+
+  test("KYUUBI #2390: RuleEliminateMarker stays in analyze phase for data masking") {
+    val db = "default"
+    val table = "src"
+    val create =
+      s"CREATE TABLE IF NOT EXISTS $db.$table (key int, value1 int) USING $format"
+    try {
+      doAs("admin", sql(create))
+      doAs("admin", sql(s"INSERT INTO $db.$table SELECT 1, 1"))
+      // scalastyle: off
+      doAs(
+        "bob", {
+          assert(sql(s"select * from $db.$table").collect() ===
+            Seq(Row(1, DigestUtils.md5Hex("1"))))
+          assert(Try(sql(s"select * from $db.$table").show(1)).isSuccess)
+        })
+    } finally {
+      doAs("admin", sql(s"DROP TABLE IF EXISTS $db.$table"))
+    }
+  }
 }
 
 class InMemoryCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {