[SPARK-51757] Fix LEAD/LAG Function Offset Exceeds Partition Size

[xin-aurora]

What changes were proposed in this pull request?

The current implementation of the prepare in OffsetWindowFunctionFrameBase:

  override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
    if (offset > rows.length) {
      fillDefaultValue(EmptyRow)
    } else {
    ...
  }

The current implementation of the write in OffsetWindowFunctionFrameBase:

   override def write(index: Int, current: InternalRow): Unit = {
    if (offset > rows.length) {
      // Already use default values in prepare.
    } else {
    ...
  }

These implementations caused the LEAD' and LAGfunctions to haveNullPointerExceptionwhen the default value is notLiteral` and the range of the default value exceeds the partition size.

This pr removes the default values in prepare and sets the default values in write.

Why are the changes needed?

Fix LEAD and LAG causes NullPointerException in the window function (SPARK-51757)

Does this PR introduce any user-facing change?

No

How was this patch tested?

Add test method in test("SPARK-51757: lead/lag column as default when offset exceeds partition") in org.apache.spark.sql.DataFrameWindowFramesSuite

Was this patch authored or co-authored using generative AI tooling?

No.

[linhongliu-db]

if the code is not needed, just remove them. and add some comments to explain the reason

[cloud-fan]

Is it the only place? Seems we should never run fillDefaultValue in prepare as the default value can be an expression that references attributes.

[beliefer]

It seems all the existing tests just cover the default value as Literal.

[beliefer]

And I think it's not related to the partition size.
The cause is apply expression on empty row.
We can check the default expression and apply fillDefaultValue(currentRow) in write if it is not a Literal, or apply it is in prepare if it is a literal.

[beliefer]

Could you check the description
The current implementation of the write in OffsetWindowFunctionFrameBase:
Where is it?