Spark computes all rows during count() on a parquet file

import org.apache.spark.SparkConf
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}

object Test extends App {
  val sparkConf = new SparkConf()
    .setAppName("test-app")
    .setMaster("local[1]")

  val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
  import sparkSession.implicits._

  val filePath = "./tempFile.parquet"
  (1 to 1000).toDF("c1")
    .repartition(10)
    .write
    .mode("overwrite")
    .parquet(filePath)

  val df = sparkSession.read.parquet(filePath)

  var rowsInHeavyComputation = 0
  def heavyComputation(row: Row): Row = {
    rowsInHeavyComputation += 1
    println(s"rowsInHeavyComputation = $rowsInHeavyComputation")
    Thread.sleep(50)
    row
  }

  implicit val encoder: ExpressionEncoder[Row] = RowEncoder(df.schema)
  val cnt = df
    .map(row => heavyComputation(row)) // map operation cannot change number of rows 
    .count()
  println(s"counting done, cnt=$cnt")
}

rowsInHeavyComputation = 1
rowsInHeavyComputation = 2
...
rowsInHeavyComputation = 999
rowsInHeavyComputation = 1000
counting done, cnt=1000