[spark] Fix RowTrackingTable reading by add _ROW_ID existing check before adding _ROW_ID (#7606)

Author: littlecoder04

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonBaseScanBuilder.scala

@@ -69,7 +69,10 @@ abstract class PaimonBaseScanBuilder
     val postScan = mutable.ArrayBuffer.empty[SparkPredicate]
 
     var newRowType = rowType
-    if (coreOptions.rowTrackingEnabled() && coreOptions.dataEvolutionEnabled()) {
+    if (
+      coreOptions.rowTrackingEnabled() && coreOptions
+        .dataEvolutionEnabled() && !rowType.containsField(SpecialFields.ROW_ID.name())
+    ) {
       newRowType = SpecialFields.rowTypeWithRowTracking(newRowType);
     }
     val converter = SparkV2FilterConverter(newRowType)

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/read/BaseScan.scala

@@ -68,7 +68,10 @@ trait BaseScan extends Scan with SupportsReportStatistics with Logging {
   val coreOptions: CoreOptions = CoreOptions.fromMap(table.options())
 
   lazy val tableRowType: RowType = {
-    if (coreOptions.rowTrackingEnabled()) {
+    if (
+      coreOptions
+        .rowTrackingEnabled() && !table.rowType().containsField(SpecialFields.ROW_ID.name())
+    ) {
       SpecialFields.rowTypeWithRowTracking(table.rowType())
     } else {
       table.rowType()

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

@@ -795,6 +795,22 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase {
     }
   }
 
+  test("Row Tracking: query row_tracking system table with filter pushdown") {
+    withTable("t") {
+      sql("CREATE TABLE t (a INT, b INT) TBLPROPERTIES ('row-tracking.enabled' = 'true')")
+      sql("INSERT INTO t VALUES (1, 10), (2, 20), (3, 30)")
+
+      val query = s"SELECT a, b FROM `t$$row_tracking` WHERE a > 1 ORDER BY a"
+      checkAnswer(sql(query), Seq(Row(2, 20), Row(3, 30)))
+
+      val scan = getScan(query)
+      assert(
+        scan.description().contains("DataFilters"),
+        s"Expected predicate pushdown (DataFilters) in scan description, but got: ${scan.description()}"
+      )
+    }
+  }
+
   test("Data Evolution: compact fields action") {
     withTable("s", "t") {
       sql("CREATE TABLE s (id INT, b INT)")
@@ -952,4 +968,5 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase {
       assert(!indexEntries.exists(entry => entry.partition().getString(0).toString.equals("p1")))
     }
   }
+
 }