[SPARK-58390][SQL] Emit row counts without deserializing Arrow payloads for empty-projection cache reads

### What changes were proposed in this pull request?

Short-circuit `ArrowCachedBatchSerializer.convertCachedBatchToInternalRow` when the projection is empty: emit `numRows` reused 0-field `UnsafeRow`s per cached batch, without deserializing or decompressing the batch's Arrow payload. The row count is already recorded on `ArrowCachedBatch`.

Also fixes the `ArrowCachedBatch` scaladoc, which listed the per-column statistics as `(upperBound, lowerBound, ...)` while both write paths produce the `ColumnStats.collectedStatistics` order `(lowerBound, upperBound, nullCount, count, sizeInBytes)`; the code was consistently lower-first everywhere, only the doc was wrong.

### Why are the changes needed?

An empty projection (e.g. a count aggregate through the row-based reader, `spark.sql.inMemoryColumnarStorage.enableVectorizedReader=false`) selects no columns, yet the reader still paid full IPC deserialization and decompression for every cached batch just to iterate its rows. That cost is pure waste: the answer is a stored integer.

### Does this PR introduce _any_ user-facing change?

No. The Arrow cache serializer (SPARK-57268) is unreleased, and the change is performance-only; results are identical.

### How was this patch tested?

Two new tests in `ArrowCachedBatchSerializerSuite`:

- `empty projection emits row counts without deserializing the Arrow payload`: hands the reader a cached batch whose Arrow payload is garbage bytes. The empty projection returns the correct number of empty rows purely from `numRows` (fails before this change with `IllegalArgumentException: capacity < 0` from the IPC reader, proving the payload used to be deserialized), while a projection that actually needs the payload still fails on the same batch, pinning that only the empty-projection case skips the read.
- `count aggregate over the cached relation with the row-based reader`: end-to-end `count(*)` over a cached relation spanning many small Arrow batches with the vectorized reader disabled, plus a `sum` over the same cached data verifying projecting reads still decode the payload correctly.

Full `ArrowCachedBatchSerializerSuite` (73 tests) and `ArrowCachedBatchKryoRegistrationSuite` pass.

### Was this patch authored or co-authored using generative AI tooling?

Yes, this pull request and its description were written by Claude Code.

Closes #57583 from viirya/arrow-cache-empty-projection.

Authored-by: Liang-Chi Hsieh <viirya@gmail.com>
Signed-off-by: Uros Bojanic <221401595+uros-b@users.noreply.github.com>

Author: viirya

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatch.scala

@@ -33,7 +33,7 @@ import org.apache.spark.sql.columnar.SimpleMetricsCachedBatch
  * The batch contains:
  *  - `numRows`: Number of rows in this batch
  *  - `arrowData`: One encapsulated Arrow RecordBatch message (with optional compression)
- *  - `stats`: Per-column statistics for partition pruning (upperBound, lowerBound, nullCount, etc.)
+ *  - `stats`: Per-column statistics for partition pruning (lowerBound, upperBound, nullCount, etc.)
  *
  * This format enables:
  *  - Zero-copy columnar reads when output is ColumnarBatch with ArrowColumnVector
@@ -42,8 +42,9 @@ import org.apache.spark.sql.columnar.SimpleMetricsCachedBatch
  *
  * @param numRows Number of rows in this cached batch
  * @param arrowData One encapsulated Arrow RecordBatch message
- * @param stats Per-column statistics as InternalRow (5 fields per column:
- *              upperBound, lowerBound, nullCount, rowCount, sizeInBytes)
+ * @param stats Per-column statistics as InternalRow (5 fields per column, in the
+ *              `ColumnStats.collectedStatistics` order:
+ *              lowerBound, upperBound, nullCount, count, sizeInBytes)
  */
 case class ArrowCachedBatch(
     numRows: Int,

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializer.scala

@@ -166,6 +166,19 @@ class ArrowCachedBatchSerializer extends SimpleMetricsCachedBatchSerializer {
       cacheAttributes: Seq[Attribute],
       selectedAttributes: Seq[Attribute],
       conf: SQLConf): RDD[InternalRow] = {
+    if (selectedAttributes.isEmpty) {
+      // Empty projection (e.g. a count aggregate over the cached relation): every cached batch
+      // already records its row count, so emit that many empty rows without touching the Arrow
+      // payload at all -- deserializing and decompressing it would be pure waste. The emitted
+      // row is a single reused 0-field UnsafeRow, matching the reuse contract of the regular
+      // path.
+      return input.mapPartitionsInternal { batchIterator =>
+        val rowWriter = new UnsafeRowWriter(0)
+        rowWriter.reset()
+        val emptyRow = rowWriter.getRow
+        batchIterator.flatMap(batch => Iterator.fill(batch.numRows)(emptyRow))
+      }
+    }
     val cacheSchema = DataTypeUtils.fromAttributes(cacheAttributes)
     val selectedSchema = DataTypeUtils.fromAttributes(selectedAttributes)
     val timeZoneId = conf.sessionLocalTimeZone

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ArrowCachedBatchSerializerSuite.scala

@@ -26,10 +26,11 @@ import org.apache.arrow.vector.{
   TimeNanoVector, TimeStampMicroTZVector, TimeStampMicroVector, TinyIntVector,
   VarBinaryVector, VarCharVector, VectorSchemaRoot, VectorUnloader}
 
-import org.apache.spark.{SparkConf, SparkUnsupportedOperationException}
+import org.apache.spark.{SparkConf, SparkException, SparkUnsupportedOperationException}
 import org.apache.spark.sql.{QueryTest, Row}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericInternalRow}
+import org.apache.spark.sql.columnar.CachedBatch
 import org.apache.spark.sql.execution.arrow.ArrowWriter
 import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
 import org.apache.spark.sql.test.{ExamplePoint, ExamplePointUDT, SharedSparkSession}
@@ -2401,6 +2402,51 @@ class ArrowCachedBatchSerializerSuite extends QueryTest with SharedSparkSession
     // If the produced root was not closed, this throws "Memory was leaked by query".
     alloc.close()
   }
+
+  test("empty projection emits row counts without deserializing the Arrow payload") {
+    // A count-style read selects no columns, and the row count is already recorded on the cached
+    // batch, so the Arrow payload must not be deserialized at all. Prove it by handing the reader
+    // a batch whose payload is garbage: the empty projection succeeds purely from numRows, while
+    // a projection that actually needs the payload fails on the same batch.
+    val serializer = new ArrowCachedBatchSerializer
+    val attrs = Seq(AttributeReference("i", IntegerType)())
+    val garbage = ArrowCachedBatch(
+      numRows = 5,
+      arrowData = Array[Byte](0x13, 0x37, 0x00, -1),
+      stats = InternalRow(null, null, 0, 5, 4L))
+    val input = spark.sparkContext.parallelize(Seq[CachedBatch](garbage), 1)
+    val conf = spark.sessionState.conf
+
+    val emptyProjection =
+      serializer.convertCachedBatchToInternalRow(input, attrs, Seq.empty, conf)
+    assert(emptyProjection.map(_.numFields).collect() === Array(0, 0, 0, 0, 0))
+
+    val fullProjection =
+      serializer.convertCachedBatchToInternalRow(input, attrs, attrs, conf)
+    intercept[SparkException] {
+      fullProjection.collect()
+    }
+  }
+
+  test("count aggregate over the cached relation with the row-based reader") {
+    // End-to-end coverage of the empty-projection read: with the vectorized reader disabled the
+    // scan produces rows via convertCachedBatchToInternalRow, and a count aggregate selects no
+    // columns. Small Arrow batches make the count span many cached batches.
+    withSQLConf(
+        SQLConf.CACHE_VECTORIZED_READER_ENABLED.key -> "false",
+        SQLConf.ARROW_EXECUTION_MAX_RECORDS_PER_BATCH.key -> "10") {
+      val df = (1 to 257).toList.toDF("i")
+      df.cache()
+      try {
+        assert(df.count() === 257)
+        checkAnswer(df.selectExpr("count(*)"), Seq(Row(257)))
+        // A projecting query over the same cached data still reads the payload correctly.
+        checkAnswer(df.selectExpr("sum(i)"), Seq(Row((1 to 257).sum.toLong)))
+      } finally {
+        df.unpersist()
+      }
+    }
+  }
 }
 
 /**