fix comments

Author: Yohahaha

fluss-spark/fluss-spark-common/src/main/scala/org/apache/fluss/spark/read/FlussPartitionReader.scala

@@ -60,7 +60,7 @@ abstract class FlussPartitionReader(
   def next0(): Boolean
 
   override def next(): Boolean = {
-    if (limit.isDefined && numRowsRead >= limit.get) {
+    if (limit.exists(numRowsRead >= _)) {
       return false
     }
     val hasNext = next0()

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkLogTableReadTest.scala

@@ -656,6 +656,14 @@ class SparkLogTableReadTest extends FlussSparkTestBase {
 
       val dfLimit = sql(s"SELECT * FROM $DEFAULT_DATABASE.t LIMIT 2")
       assert(flussAppendScans(dfLimit).flatMap(_.limit).distinct == Seq(2))
+
+      // Verify limit pushdown actually reduces rows read via metrics
+      dfLimit.collect()
+      val batchScanExec = dfLimit.queryExecution.executedPlan.collectFirst {
+        case b: BatchScanExec => b
+      }.get
+      val numRowsRead = batchScanExec.metrics(FlussMetrics.NUM_ROWS_READ).value
+      assert(numRowsRead == 2L, s"Expected 2 rows read with limit pushdown, got $numRowsRead")
     }
   }
 }

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/SparkPrimaryKeyTableReadTest.scala

@@ -453,8 +453,16 @@ class SparkPrimaryKeyTableReadTest extends FlussSparkTestBase {
       val dfNoLimit = sql(s"SELECT * FROM $DEFAULT_DATABASE.t")
       assert(flussUpsertScan(dfNoLimit).flatMap(_.limit).isEmpty)
 
-      val dfLimit = sql(s"SELECT * FROM $DEFAULT_DATABASE.t LIMIT 2")
-      assert(flussUpsertScan(dfLimit).flatMap(_.limit).contains(2))
+      val dfLimit = sql(s"SELECT * FROM $DEFAULT_DATABASE.t WHERE dt = '2026-01-01' LIMIT 1")
+      assert(flussUpsertScan(dfLimit).flatMap(_.limit).contains(1))
+
+      // Verify limit pushdown actually reduces rows read via metrics
+      dfLimit.collect()
+      val batchScanExec = dfLimit.queryExecution.executedPlan.collectFirst {
+        case b: BatchScanExec => b
+      }.get
+      val numRowsRead = batchScanExec.metrics(FlussMetrics.NUM_ROWS_READ).value
+      assert(numRowsRead == 1L, s"Expected 1 rows read with limit pushdown, got $numRowsRead")
     }
   }
 

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakeLogTableReadTest.scala

@@ -20,8 +20,10 @@ package org.apache.fluss.spark.lake
 import org.apache.fluss.config.{ConfigOptions, Configuration}
 import org.apache.fluss.metadata.DataLakeFormat
 import org.apache.fluss.spark.SparkConnectorOptions.BUCKET_NUMBER
+import org.apache.fluss.spark.read.FlussMetrics
 
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 
 import java.nio.file.Files
 
@@ -547,6 +549,14 @@ abstract class SparkLakeLogTableReadTest extends SparkLakeTableReadTestBase {
 
       val df = sql(s"SELECT * FROM $DEFAULT_DATABASE.t_union_limit LIMIT 2")
       assert(flussScan(df).flatMap(_.limit).distinct == Seq(2))
+
+      // Verify limit pushdown actually reduces rows read via metrics
+      df.collect()
+      val batchScanExec = df.queryExecution.executedPlan.collectFirst {
+        case b: BatchScanExec => b
+      }.get
+      val numRowsRead = batchScanExec.metrics(FlussMetrics.NUM_ROWS_READ).value
+      assert(numRowsRead == 2L, s"Expected 2 rows read with limit pushdown, got $numRowsRead")
     }
   }
 

fluss-spark/fluss-spark-ut/src/test/scala/org/apache/fluss/spark/lake/SparkLakePrimaryKeyTableReadTestBase.scala

@@ -20,8 +20,10 @@ package org.apache.fluss.spark.lake
 import org.apache.fluss.config.{ConfigOptions, Configuration}
 import org.apache.fluss.metadata.DataLakeFormat
 import org.apache.fluss.spark.SparkConnectorOptions.{BUCKET_NUMBER, PRIMARY_KEY}
+import org.apache.fluss.spark.read.FlussMetrics
 
 import org.apache.spark.sql.Row
+import org.apache.spark.sql.execution.datasources.v2.BatchScanExec
 
 import java.nio.file.Files
 
@@ -481,6 +483,14 @@ abstract class SparkLakePrimaryKeyTableReadTestBase extends SparkLakeTableReadTe
       val query =
         sql(s"SELECT id, score FROM $DEFAULT_DATABASE.t_pk_union_limit LIMIT 2")
       assert(flussScan(query).flatMap(_.limit).distinct == Seq(2))
+
+      // Verify limit pushdown actually reduces rows read via metrics
+      query.collect()
+      val batchScanExec = query.queryExecution.executedPlan.collectFirst {
+        case b: BatchScanExec => b
+      }.get
+      val numRowsRead = batchScanExec.metrics(FlussMetrics.NUM_ROWS_READ).value
+      assert(numRowsRead == 2L, s"Expected 2 rows read with limit pushdown, got $numRowsRead")
     }
   }