[VL] Add min/max partition stats to columnar InMemoryRelation cache for partition pruning

[yaooqinn]

What changes were proposed in this pull request?

This PR fills the long-standing TODO in ColumnarCachedBatchSerializer.buildFilter
by adding native-side min/max partition statistics to the columnar InMemoryRelation
cache, enabling partition-level pruning when filtering over a cached DataFrame.

The Velox backend now computes per-batch min/max/null-count stats during
serializeWithStats (cpp side, Arrow C-Data export-free) and embeds them
in the cached batch envelope. On the read path, the JVM serializer extends
SimpleMetricsCachedBatchSerializer and routes batches with stats through
the inherited buildFilter for partition pruning; batches without stats
(legacy v1 binary, or the SQLConf gate disabled) lazy-pass-through unchanged.

Type matrix covered (vanilla Spark ColumnStats.scala parity):

• Integer family: TINYINT / SMALLINT / INT / BIGINT
• Date / Timestamp / TimestampNTZ
• YearMonth / DayTime intervals
• Decimal (short P<19 / long P>=19 / HUGEINT)
• String (JVM marshal; cpp VARCHAR scan deferred — see notes)
• Boolean
• Float / Double (NaN-poison guarded)

VARCHAR cpp-side scan is intentionally deferred (supported=0 from cpp,
no JVM crash path) due to Velox StringView lifetime considerations across
RowVector boundaries — tracked for a follow-up PR.

Why are the changes needed?

Partition stats let InMemoryRelation skip whole cached partitions that
cannot satisfy a filter predicate, materially speeding up repeated point
lookups and selective filters over a cached DataFrame — a common pattern
in iterative SQL workloads (notebooks, ML feature engineering, BI tools
with cached intermediate results). Vanilla Spark has had this since
SPARK-32274 (3.1, ~5 years); Gluten's columnar cache previously
extended CachedBatchSerializer directly, leaving the buildFilter TODO
empty and skipping pruning entirely.

Does this PR introduce any user-facing change?

Yes — a new SQLConf:

Key	Default	Meaning
spark.gluten.sql.columnar.tableCache.partitionStats.enabled	false	Enable native partition stats computation + read-side pruning

Default-off ship for the first release; flip to default-on after community
feedback on the included benchmark + a release of soak time. Gate is
double-checked on both the write path (skip stats compute) and read path
(via the inherited buildFilter).

How was this patch tested?

Unit / integration tests:

• 19 cpp gtest cases (Velox velox_operators_test) — per-type min/max scan
  • framing + carry-overflow + NaN poison guard + truncate semantics
• 8 JVM Scala suites, 37 cases — Kryo wire format / stats blob marshal /
  buildFilter wrapper / per-arm prune semantics (BIGINT / Date / Timestamp /
  String / NaN sentinel / numCols cap)
• 1 Velox e2e suite (ColumnarCachedBatchE2ESuite) — cache + equality
  filter end-to-end, plan shape verification, numOutputRows prune
  evidence

Benchmark (backends-velox/benchmarks/ColumnarTableCachePartitionStatsBenchmark-results.txt):

100M-row range partitioned by c2 into 32 in-memory partitions,
groupBy + sum/count/avg follow-up to make pruned-partition savings
visible. Driver: local[1], AMD EPYC 7763.

Case	partitionStats off	partitionStats on	Speedup
cache build (write path)	126,425 ms	131,431 ms	1.0x (no measurable overhead)
filter+agg, high selectivity (c2 < 1000, ~0.001%)	4,431 ms	1,744 ms	2.5x
filter+agg, low selectivity (c2 < 50000000, ~50%)	5,332 ms	3,392 ms	1.6x
filter+agg, point lookup (c2 = 50000000, 1 row)	4,343 ms	1,686 ms	2.6x

Write path shows no measurable overhead — stats computation runs in cpp
inside the existing batch scan loop.

Notes on commit history

This PR retains the atomic per-slice commit history (~58 commits) rather
than squashing, to make per-feature git bisect and incremental review
practical. Each commit is self-contained: a feature slice, its test, and
any fixes amended into the slice.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[weiting-chen]

Thanks for this excellent PR — the partition stats design is clean and well-tested. A few suggestions for hardening (none are merge-blocking).

[Copilot]

Pull request overview

This PR adds native-produced per-partition min/max/null-count stats to the Velox-backed columnar InMemoryRelation cache so Spark can apply partition pruning when filtering cached DataFrames (gated by a new SQLConf).

Changes:

• Introduces spark.gluten.sql.columnar.tableCache.partitionStats.enabled (default false) to control stats collection and pruning.
• Extends the Velox serializer/JNI to emit a framed payload [magic | statsLen | statsBlob | bytesLen | bytesBlob] and adds JVM-side parsing + Kryo (de)serialization for cached batches carrying stats.
• Adds extensive C++ and Scala test coverage plus a benchmark demonstrating pruning benefits.

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala	Adds the new SQLConf flag for enabling cache partition stats.
gluten-arrow/src/main/java/org/apache/gluten/vectorized/ColumnarBatchSerializerJniWrapper.java	Adds JNI serializeWithStats and a “capability check” helper (needs correction).
docs/Configuration.md	Documents the new configuration key.
cpp/core/operators/serializer/ColumnarBatchSerializer.h	Adds a virtual framedSerializeWithStats hook (default empty) for backend override.
cpp/velox/operators/serializer/VeloxColumnarBatchSerializer.h	Declares stats computation and framed serialization for Velox.
cpp/velox/operators/serializer/VeloxColumnarBatchSerializer.cc	Implements min/max scan + NaN guard and framed serialization (needs length-range guarding).
cpp/core/jni/JniWrapper.cc	Exposes serializeWithStats JNI entry point returning framed bytes.
cpp/velox/tests/VeloxColumnarBatchSerializerTest.cc	Adds detailed gtests for stats correctness and wire layout.
backends-velox/src/main/scala/org/apache/spark/sql/execution/ColumnarCachedBatchSerializer.scala	Adds CachedColumnarBatch w/ stats+schema, Kryo framing/stat marshaling, and buildFilter wrapper (needs robust fallback handling).
backends-velox/src/test/scala/org/apache/spark/sql/execution/*.scala	Adds JVM unit + e2e tests for framing/marshal/pruning behavior.
backends-velox/src/test/scala/org/apache/spark/sql/execution/benchmark/ColumnarTableCachePartitionStatsBenchmark.scala	Adds benchmark harness to measure overhead and pruning benefit.
benchmarks/ColumnarTableCachePartitionStatsBenchmark-results.txt	Adds benchmark output snapshot.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[liuneng1994]

Thanks for the substantial work and the thorough TDD trail. Reviewed focusing on Spark behavior parity and test coverage. Three findings can return wrong results / crash and aren't covered by tests; flagging before any future default-on flip.

🔴 1. Non-binary collation strings can be wrongly pruned (silent wrong results)

CachedColumnarBatchKryoSerializer.isDispatchable accepts any StringType (the comment even says "any collation"):

case _: StringType => true // truncated to 256B; see encodeStringBounds (any collation)

But both the cpp side (scanMinMax<StringView> → memcmp) and JVM encodeStringBounds (byte-wise truncate + +1 carry) compute min/max in unsigned byte order. For non-binary collations (UTF8_LCASE, UNICODE_CI, …) Spark's equality / ordering is not byte order, so a partition whose true min/max under the collation overlaps the predicate range may have a byte-order min/max that does not — the partition gets pruned and matching rows disappear.

Suggested fix: gate on the binary collation only, e.g.

case s: StringType if s.collationId == CollationFactory.UTF8_BINARY_COLLATION_ID => true

and add an e2e test under a non-binary collation.

🔴 2. CONSTANT/DICTIONARY-encoded columns + IsNull can prune wrongly

if (child == nullptr || !child->isFlatEncoding()) { continue; }   // computeStats

For dict/const-encoded children, nullCount stays at 0 and we then emit nullCount=0, count=numRows, supported=0. The vanilla SimpleMetricsCachedBatchSerializer.buildFilter evaluates IsNull(col) purely from nullCount (it does not require supported/lower/upper) — so it concludes "no nulls in this partition" and prunes it. If the column is in fact a CONSTANT-null vector (very common after outer joins or projection of literal null), IsNull(col) will incorrectly return zero rows.

Suggested fix: at minimum decode nullCount for non-flat encodings (e.g. via DecodedVector or child->mayHaveNulls() + per-row check), or skip writing the nullCount slot so the parent treats it as unknown. New test needed: cache a frame containing an all-null column on the non-flat path, then filter with col.isNull and assert the count equals N (not 0).

🟠 3. supportsStatsExt() does not actually probe the cpp symbol

public static boolean supportsStatsExt() {
  ...
  ColumnarBatchSerializerJniWrapper.class.getDeclaredMethod("serializeWithStats", long.class);
  ...
}

This reflects on the JVM class's own declared method, which is always true once the PR lands. The comment acknowledges "the cpp symbol is verified at first real invocation; callers must catch UnsatisfiedLinkError there too" — but convertColumnarBatchToCachedBatch does not catch. So a new gluten jar paired with an older libgluten.so will hard-crash with UnsatisfiedLinkError instead of falling back to the legacy serialize() path that the comment promises.

Suggested fix: try/catch UnsatisfiedLinkError once around an inert probe call (or do dlsym natively at lib-load), cache the result, and only then dispatch to serializeWithStats.

---

🟡 Smaller correctness / parity notes

• NaN poison preserves partial nullCount. scanMinMax returns false on the first NaN; the nullCnt accumulated up to that point is still written into stats. It happens to be conservative for IsNull/IsNotNull (always under-counts), but it is a behavior divergence from ColumnStats.gatherNullStats and is fragile to future predicate-eval changes — recommend zeroing nullCount when the column is NaN-poisoned.
• TimestampNTZ. JVM dispatch lumps TimestampType | TimestampNTZType into the 8B Long arm; cpp only inspects TypeKind::TIMESTAMP. Worth verifying Velox's NTZ TypeKind in Gluten's mapping and adding an NTZ e2e — currently no NTZ test.
• buildFilter lazy-split calls parent(index, runIt) per contiguous run. Each invocation may compile a fresh codegen filter; in mixed (stats / no-stats) iterators this can balloon. Consider memoizing or pre-warming the parent closure.

---

Test coverage gaps to add before default-on

The unit / wire tests are thorough, but the following e2e gaps should be closed before flipping the default to true:

1. Non-binary collation string column + equality / range filter (covers #1).
2. CONSTANT/dict-encoded all-null column + col.isNull filter (covers #2).
3. Stale-symbol fallback: simulate serializeWithStats missing on the cpp side and assert graceful fallback to the legacy path (covers #3).
4. TimestampNTZ end-to-end equality.
5. Cross-config: cache built with stats=on, read after toggling stats=off (and vice versa); also mixed v1-binary + v2-stats batches in the same partition.
6. Disk spill via Kryo: existing Kryo tests round-trip in-memory only; cover the DiskStore spill + read-back path with stats present.
7. Mixed supported=1 and supported=0 columns in one batch — wire alignment is only implicitly tested.
8. Decimal with scale > 0 and long-Decimal (precision > 18) end-to-end.
9. YearMonth / DayTime interval end-to-end (only marshal unit tests today).
10. Schema mismatch: stats column count vs schema.length mismatch — currently relies on caller invariant with no require.

---

Non-blocking polish

• The production code carries a lot of TDD-process noise (PA-3.2 RED, PA-6.A, refs: todos/features/…). These are valuable in the commit messages but are likely to confuse future maintainers when they appear in // comments of merged code — consider rewriting them as forward-looking design comments at squash / merge.
• serializeStats(stats, schema) accepts schema == null and falls back to BIGINT; the write path always passes a non-null schema, but no require enforces it. A future caller could regress silently — recommend require(schema != null, …) once V1-only paths are gone.
• count / nullCount are uint32 on the wire — fine in practice, but a comment about the >2³¹-row assumption (or widening to uint64) would help.
• cpp framedSerializeWithStats writes pushU64(0) for per-column sizeInBytes; a one-line comment that this is intentional because CachedColumnarBatch.sizeInBytes is overridden at the batch level would prevent future "fix me" PRs.

---

Overall the design is clean and the TDD discipline is great; default-off ship is the right call. With #1–#3 fixed and the e2e gaps closed, this should be solid for a default-on flip.

[jackylee-ch]

@yaooqinn it looks similar to the pr(#12073 ) I submitted before, both of which implement filter pushdown for the cache, with only slight differences in implementation details.

[jackylee-ch]

it seems the current pr also implement the CachedBatch skipping with pushed filter not just the partition, right?

[jackylee-ch]

here is the main different between this two prs.

[zhli1142015]

I reviewed the partition-stats changes and left a few correctness comments.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86