[GLUTEN-11915][VL] Support RowBasedChecksum in ColumnarShuffleWriter (SPARK-51756)

[jaylisde]

Summary

Spark 4.1 introduced RowBasedChecksum (SPARK-51756) for detecting non-deterministic stage retries. When spark.sql.shuffle.orderIndependentChecksum.enabled or spark.sql.shuffle.checksum.mismatchFullRetry.enabled is true, the shuffle writer must compute an order-independent per-row checksum and pass it via MapStatus.checksumValue to the driver for comparing across task attempts.

Problem: Gluten's ColumnarShuffleWriter always returns checksumValue = 0, causing the driver to skip non-deterministic retry detection. If a task retry produces different output (e.g., due to round-robin partitioning), downstream consumers may silently read inconsistent data without triggering a full stage retry.

Fix: Implement native C++ row-based checksum computation in VeloxHashShuffleWriter. For each row in doSplit(), serialize via UnsafeRowFast and compute XXH64 hash. Aggregate per-partition using XOR+SUM (order-independent). Return checksum array via JNI to Scala layer, which passes the aggregated value to MapStatus.checksumValue.

Changes

• VeloxHashShuffleWriter.cc: Added computeRowBasedChecksums() using UnsafeRowFast + XXH64 with per-partition XOR+SUM aggregation.
• Options.h, ShuffleWriter.h/cc: Added rowBasedChecksumEnabled option and rowBasedChecksums() accessor.
• JniWrapper.cc: Accept boolean config param, return checksum array.
• GlutenSplitResult.java, ShuffleWriterJniWrapper.java: Added rowBasedChecksums field and param.
• ColumnarShuffleWriter.scala: Read SQLConf (OR logic), pass to native, use for MapStatus.
• GlutenMapStatusUtil.scala (shims/spark33-41): Cross-version MapStatus compatibility.
• RowBasedChecksumTest.cc: C++ unit test for order-independence, null handling, determinism.
• GlutenMapStatusEndToEndSuite.scala: Integration test with ansiFallback=false.

Test

• C++ unit test: 4/4 pass (order-independence, data-change detection, null handling, deterministic)
• GlutenMapStatusEndToEndSuite: 3/3 pass (propagation, deterministic, data-change detection)

Partially addresses #11915.

Note

File-based shuffle checksum (.checksum file with ADLER32 for corruption diagnosis) will be addressed in a follow-up PR.

[wForget]

Spark 4.1 introduced shuffle checksum end-to-end verification (SPARK-53322), requiring MapStatus.checksumValue to be non-zero and .checksum files to contain valid per-partition checksums.

Do you mean https://issues.apache.org/jira/browse/SPARK-51756?

[jaylisde]

Thanks @wForget for catching that. The correct reference should be SPARK-54663. I'll update the PR title and description.

[wForget]

Thanks @wForget for catching that. The correct reference should be SPARK-54663. I'll update the PR title and description.

SPARK-54663 proposes row-based checksum, but the current implementation is based on data file.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[jaylisde]

Thanks @wForget. Updated to proper RowBasedChecksum (SPARK-51756) with native per-row XXH64 + order-independent aggregation. File-based checksum will be a follow-up PR.

[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

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[philo-he]

@marin-ma, could you take a look when you get a chance?

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86