[VL] Add cross-config / cross-build-cycle invariant tests for ColumnarCachedBatchSerializer (#12124)

Extend ColumnarCachedBatchE2ESuite with three lifecycle invariant tests that
exercise the cached-batch wire format across SQLConf transitions:

  1. cross-config: build with stats=true, read with stats=false
     -- wire format is build-time-decided; v2-with-stats payload must
        survive a reader-time downgrade and prune must still engage.
  2. cross-config (reverse): build with stats=false, read with stats=true
     -- legacy v1 payload (stats=null) at build time; reader must NOT
        fabricate stats and must fall back to full scan.
  3. cross-build-cycle: same logical query rebuilt twice with different
     stats settings (round 1 stats=true, round 2 stats=false). Round 2
     must re-honor stats=false; the serializer must not reuse stale gate
     state from round 1.

Each test asserts (R-correct) result row count, (R-path) the cached
batches are served by ColumnarCachedBatchSerializer (not vanilla
DefaultCachedBatchSerializer), and (R-prune, when expectPrune=true)
that InMemoryTableScanExec.numOutputRows reflects partition pruning.

The shared assertion logic is factored into a private helper
`assertGlutenCachedPlanAndPrune(df, expectPrune)`. Its scaladoc
documents an intentional asymmetry: the reverse "no prune" direction
is not observable via numOutputRows on the Gluten native path (the same
baseline test "numOutputRows reflects post-filter row count" already
notes that outRows may legitimately be 0 even with full pruning, because
the surviving row is delivered through the native scan metrics path).
The expectPrune=false branch therefore intentionally performs path-only
verification.

All 15 suite tests pass locally (Spark 3.3, Velox backend).

Generated-by: Claude claude-opus-4.7

Author: yaooqinn

backends-velox/src/test/scala/org/apache/spark/sql/execution/ColumnarCachedBatchE2ESuite.scala

@@ -74,6 +74,26 @@ class ColumnarCachedBatchE2ESuite
       .cache()
   }
 
+  // Caller must have triggered execution so numOutputRows is populated.
+  // expectPrune=false is path-only: numOutputRows is unreliable for "no prune"
+  // on the Gluten native path (surviving rows bypass the IMS counter, see
+  // baseline "numOutputRows reflects post-filter row count" test).
+  private def assertGlutenCachedPlanAndPrune(df: DataFrame, expectPrune: Boolean): Unit = {
+    val plan = df.queryExecution.executedPlan
+    val ims = find(plan) {
+      case _: InMemoryTableScanExec => true
+      case _ => false
+    }
+      .get.asInstanceOf[InMemoryTableScanExec]
+    val serName = ims.relation.cacheBuilder.serializer.getClass.getSimpleName
+    assert(serName == "ColumnarCachedBatchSerializer", s"got $serName")
+    if (expectPrune) {
+      val outRows = ims.metrics("numOutputRows").value
+      val upperBound = (N / P) * 2
+      assert(outRows <= upperBound, s"numOutputRows=$outRows > $upperBound (N=$N, P=$P)")
+    }
+  }
+
   test("e2e cache + equality filter: no crash + correct result") {
     val cached = cacheRange()
     try {
@@ -385,4 +405,87 @@ class ColumnarCachedBatchE2ESuite
       }
     }
   }
+
+  // Cross-config: build with stats enabled, read with stats disabled.
+  // Wire format is build-time-decided, so reader-time SQLConf must not affect prune.
+  test("cross-config: build with stats enabled, read with stats disabled") {
+    var cached: DataFrame = null
+    var filtered: DataFrame = null
+    var result: Long = -1L
+    withSQLConf(
+      GlutenConfig.COLUMNAR_TABLE_CACHE_PARTITION_STATS_ENABLED.key -> "true") {
+      val df = cacheRange()
+      df.count()
+      cached = df
+    }
+    try {
+      withSQLConf(
+        GlutenConfig.COLUMNAR_TABLE_CACHE_PARTITION_STATS_ENABLED.key -> "false") {
+        filtered = cached.filter(col("k") === pivot)
+        result = filtered.count()
+      }
+      assert(result == 1L, s"got $result")
+      assertGlutenCachedPlanAndPrune(filtered, expectPrune = true)
+    } finally {
+      cached.unpersist()
+    }
+  }
+
+  // Reverse: legacy v1 payload at build (stats=null), reader cannot fabricate
+  // stats. Distinct from the same-config legacy test: this forces cross-config.
+  test("cross-config: build with stats disabled, read with stats enabled") {
+    var cached: DataFrame = null
+    var filtered: DataFrame = null
+    var result: Long = -1L
+    withSQLConf(
+      GlutenConfig.COLUMNAR_TABLE_CACHE_PARTITION_STATS_ENABLED.key -> "false") {
+      val df = cacheRange()
+      df.count()
+      cached = df
+    }
+    try {
+      withSQLConf(
+        GlutenConfig.COLUMNAR_TABLE_CACHE_PARTITION_STATS_ENABLED.key -> "true") {
+        filtered = cached.filter(col("k") === pivot)
+        result = filtered.count()
+      }
+      assert(result == 1L, s"got $result")
+      assertGlutenCachedPlanAndPrune(filtered, expectPrune = false)
+    } finally {
+      cached.unpersist()
+    }
+  }
+
+  // Round 2 must re-honor the new SQLConf, not reuse stale gate decision /
+  // payload from round 1.
+  test("cross-build-cycle: unpersist + toggle stats config + rebuild same query") {
+    var resultA: Long = -1L
+    var resultB: Long = -1L
+    withSQLConf(
+      GlutenConfig.COLUMNAR_TABLE_CACHE_PARTITION_STATS_ENABLED.key -> "true") {
+      val df = cacheRange()
+      try {
+        df.count()
+        val filtered = df.filter(col("k") === pivot)
+        resultA = filtered.count()
+        assert(resultA == 1L, s"round 1: got $resultA")
+        assertGlutenCachedPlanAndPrune(filtered, expectPrune = true)
+      } finally {
+        df.unpersist(blocking = true)
+      }
+    }
+    withSQLConf(
+      GlutenConfig.COLUMNAR_TABLE_CACHE_PARTITION_STATS_ENABLED.key -> "false") {
+      val df = cacheRange()
+      try {
+        df.count()
+        val filtered = df.filter(col("k") === pivot)
+        resultB = filtered.count()
+        assert(resultB == 1L, s"round 2: got $resultB")
+        assertGlutenCachedPlanAndPrune(filtered, expectPrune = false)
+      } finally {
+        df.unpersist(blocking = true)
+      }
+    }
+  }
 }