Fix a special case in limit where it could return an empty batch with the wrong number of columns (#13938)

[res-life]

Picked from main branch(#13938)

PR #13903 (target branch 25.12) needs to use the CSV files in #13938

[res-life]

build

[greptile-apps]

Greptile Overview

Greptile Summary

Fixes a bug where GpuBaseLimitIterator could return an empty batch with the wrong number of columns when the offset exceeded all available rows in a partition.

• Root cause: The original code captured numCols from the first input batch, but when all batches were consumed during offset processing and an empty batch needed to be returned, it created an incorrect empty batch using ArrayBuffer[GpuColumnVector](numCols) which had no actual column data types
• The fix: Passes the expected output dataTypes array to GpuBaseLimitIterator and uses GpuColumnVector.emptyBatch(StructType(fields)) to create properly-typed empty batches
• Impact: This bug manifested when the empty batch was broadcast (e.g., in cross joins with an empty side), causing column count mismatches
• Testing: Added integration test test_empty_cross_side_with_limit that reproduces the scenario with a cross join between an empty table and a non-empty table with different schemas

Confidence Score: 5/5

• This PR is safe to merge - it's a targeted bug fix with proper test coverage
• The fix is well-scoped, addressing a specific edge case in limit processing. The change follows established patterns in the codebase (passing dataTypes array), includes both unit and integration tests, and the code logic is straightforward and correct.
• No files require special attention

Important Files Changed

File Analysis

Filename	Score	Overview
sql-plugin/src/main/scala/com/nvidia/spark/rapids/limit.scala	5/5	Core fix: Added dataTypes parameter to GpuBaseLimitIterator to properly construct empty batches with correct schema instead of deriving from input batch
tests/src/test/scala/com/nvidia/spark/rapids/LimitRetrySuite.scala	5/5	Updated test to pass new dataTypes parameter to GpuBaseLimitIterator constructor, also updated copyright year
integration_tests/src/main/python/join_test.py	5/5	Added new integration test test_empty_cross_side_with_limit that reproduces the bug with cross join of empty table
integration_tests/src/test/resources/t0.csv	5/5	Test data file with header-only CSV (two columns: c0, c1) representing empty table for reproducing the bug
integration_tests/src/test/resources/t1.csv	5/5	Test data file with one column (c0) and one row (true) for cross join test scenario

Sequence Diagram

sequenceDiagram
    participant Client as Spark Client
    participant Limit as GpuBaseLimitIterator
    participant Batch as Input Batch Iterator
    participant Empty as GpuColumnVector.emptyBatch

    Client->>Limit: next()
    Limit->>Batch: input.next()
    Batch-->>Limit: batch (may be null or empty)
    
    alt offset >= all rows in partition
        Note over Limit: All batches consumed during offset skip
        Limit->>Empty: emptyBatch(StructType(fields))
        Note over Empty: Creates empty batch with<br/>correct schema from dataTypes
        Empty-->>Limit: ColumnarBatch with correct columns
    else normal case
        Limit->>Limit: slice batch
        Limit-->>Client: ColumnarBatch with data
    end
    
    Limit-->>Client: ColumnarBatch

Loading

[greptile-apps]

_{3 files reviewed, no comments}

_{Edit Code Review Agent Settings | Greptile}

[firestarman]

LGTM since this is just a cherry-pick.