Fix PCBS binary column decoding [serial-ut] [databricks]

[wjxiz1992]

JaCoCo production line coverage: +33 lines (sql-plugin +33; fix-line intersection, shim 330, vs c5e81213e; validated source 8039820e1)

Fixes #15613.

Description

This is a correctness fix for GPU queries that read Spark BinaryType columns from the Parquet-backed cache. With the Parquet Cached Batch Serializer (PCBS), the same cached query returned the expected binary value on CPU but failed on GPU because the read path did not convert the Parquet string representation back to Spark's binary representation. This PR adds that conversion for top-level and nested binary columns so the cache scan can run on the GPU with CPU-equivalent results. Other Parquet reads and cached data without binary columns are unchanged.

What changed

• Reuse the existing Parquet schema-evolution conversion to reconstruct binary values while PCBS decodes a cached batch.
• Apply the conversion recursively to nested binary fields and enable BinaryType for GPU in-memory table scans.
• Run decoding and conversion inside the existing no-split retry boundary, closing partially converted columns before retrying.
• Avoid a redundant device copy when the conversion already returns an owning top-level column.

The regression test covers non-empty, empty, and null binary values at the top level, inside a struct, and in arrays and maps (including empty/null collections and empty/null elements), both vectorized-reader settings, injected OOM, CPU/GPU equality, and an explicit GpuInMemoryTableScanExec plan assertion.

Review focus

Please focus on resource ownership, retry cleanup, and nested-column reconstruction in the shared string-to-binary helper.

AI assistance: The change and PR description were prepared with Codex assistance.

Validation

Validated the source committed in 8039820e1 with Spark 3.3.0, Scala 2.12, Python 3.10.18 and an RTX 5880 Ada GPU:

• CachedBatchWriterSuite: Tests: succeeded 8, failed 0, canceled 0, ignored 0, pending 0; Maven BUILD SUCCESS in 3:09.
• Python selection test_cache_binary_on_gpu or (test_cache_cpu_gpu_mixed and Decimal): 16 passed, 32469 deselected in 19.45 seconds. The binary cases retain injected OOM, both vectorized-reader settings, CPU/GPU equality and the GpuInMemoryTableScanExec assertion; the remaining cases check decimal cache compatibility.
• mvn package -pl dist,integration_tests -am -DskipTests -Dbuildver=330: all 18 selected modules succeeded; BUILD SUCCESS in 1:20. Both Maven runs used the worktree-local repository and the required GPU allocation flags.
• JaCoCo covered 33 of 60 added JVM production lines against c5e81213e. All affected JVM production code is in sql-plugin: ColumnCastUtil 3, GpuExecOverrides 3, ParquetCachedBatchSerializer 20, and ParquetSchemaUtils 7. The 18 affected class-family files were verified byte-for-byte against the runtime JAR and analyzed without class-ID mismatch.
• All 37 versioned support snapshots and the generic snapshot report BINARY=S for InMemoryTableScanExec.

Performance and CI detail

A local A/B check compared the ownership-dispatch simplification with the preceding production runtime (dd55be18d; the intervening 4c6437004 changes only the Python fixture). Each iteration scanned 4,000,000 cached rows with top-level and nested 128-byte binary values, with 5 warmups and 12 measured iterations. Both plans used the GPU cache scan. Median latency was 47.78 ms before and 37.50 ms after; mean latency was 47.39 ms and 37.19 ms. This single local comparison showed no slowdown; it does not establish a causal speedup, especially because the baseline overlapped a CPU build. The change preserves the same reference-count and copy operations while simplifying dispatch.

[serial-ut] is used because Blossom #14497 aborted in Spark 4.0.1/Scala 2.13 GpuDeviceManagerSuite during parallel execution. Its RMM pool size test failed on its first 1.520384 GiB allocation with Maximum pool size exceeded, before the expected second-allocation OOM assertion. The test sizes that allocation from free VRAM sampled before Spark initializes its pool; concurrent workers can change that free-memory value. The other four CI lanes were canceled by fail-fast. Serial execution removes that worker contention while retaining the test matrix.

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
  (Please provide the names of the existing tests in the PR description.)
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Confidence Score: 5/5

The PR appears safe to merge, with no outstanding actionable findings.

Findings

1.  Versioned Support Metadata Is Stale ▶

Summary

• Converts Parquet string-backed values into Spark-compatible binary columns recursively through structs, arrays, and maps.
• Moves decoding and conversion into a no-split OOM retry boundary with cleanup of partially converted columns.
• Enables binary input for GPU in-memory table scans and updates support metadata across Spark versions.
• Adds CPU/GPU parity coverage for null, empty, top-level, and nested binary values under vectorized-reader and injected-OOM configurations.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Parquet cached batch] --> B[Read cuDF table]
  B --> C{Spark schema requires conversion?}
  C -->|STRING-backed BinaryType| D[Reconstruct LIST of UINT8]
  C -->|Negative-scale decimal| E[Decimal bit cast]
  C -->|No conversion| F[Retain column]
  D --> G[Recursively rebuild nested columns]
  E --> G
  F --> G
  G --> H[GpuColumnVector batch]
  B -. retryable allocation failure .-> I[No-split retry and cleanup]
  G -. retryable allocation failure .-> I
  I --> B

Loading

_{Reviews (8) · Last reviewed commit: "Simplify ownership dispatch for converte..."}

[Copilot]

🟡 Changes recommended

The new PCBS decode path introduces an avoidable double device-side copy for STRING→Binary conversions, which can add unnecessary GPU memory/latency overhead.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Fixes GPU cached-batch (PCBS) decoding for Spark BinaryType columns that were serialized as Parquet strings, ensuring decoded cuDF STRING columns are converted back to LIST<UINT8> before being exposed to Spark as binary (including nested fields).

Changes:

• Extracted a reusable STRING→Binary (LIST) reconstruction helper in ParquetSchemaUtils.
• Applied deep, recursive STRING→Binary conversion during PCBS decode for both top-level and nested BinaryType columns.
• Added a regression test covering non-empty, empty, and null binary values at top-level and within a nested struct.

File summaries

File	Description
tests/src/test/scala/com/nvidia/spark/rapids/CachedBatchWriterSuite.scala	Adds a PCBS round-trip regression test for top-level and nested BinaryType.
sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/ParquetSchemaUtils.scala	Factors out STRING→Binary reconstruction into convertStringToBinary.
sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/ParquetCachedBatchSerializer.scala	Uses deep conversion to fix PCBS decode for STRING-backed BinaryType (including nested).

Review details

• Files reviewed: 3/3 changed files
• Comments generated: 1
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[wjxiz1992]

Updated.

[wjxiz1992]

build

[greptile-apps]

Versioned Support Metadata Is Stale

The common GpuExecOverrides code now supports top-level and nested binary columns for InMemoryTableScanExec, but only the generic and Spark 3.3.0 capability files were updated. Other supported version snapshots, including Spark 3.4.0 and 4.2.0, still report binary columns as unsupported. This non-blocking inconsistency gives documentation and capability consumers conflicting support information, so the remaining versioned metadata should be regenerated.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

[wjxiz1992]

The remaining versioned snapshots were regenerated in 81731cb. I verified that all 37 versioned snapshots and the generic snapshot now report BINARY=S for InMemoryTableScanExec.

[wjxiz1992]

build

[wjxiz1992]

build

[wjxiz1992]

build

[thirtiseven]

LGTM

[firestarman]

One NIT, but i am fine with merging it as-is.

[firestarman]

NIT: The original tmp !=cv assumes tmp to be a ColumnView, but it is no longer always true.
We can simplify this part as follows:

convertTypeAToTypeB(cv, dataType, predicate, toClose) match {
  case vector: ColumnVector => vector.incRefCount()
  case _ => tmp.copyToColumnVector()
}

[wjxiz1992]

Thanks, updated.

[wjxiz1992]

build

[wjxiz1992]

build