[SPARK-45430] Fix for FramelessOffsetWindowFunction when IGNORE NULLS and offset > rowCount

[vitaliili-db]

What changes were proposed in this pull request?

This is a fix for the failure when function that utilized FramelessOffsetWindowFunctionFrame is used with ignoreNulls = true and offset > rowCount.

e.g.

select x, lead(x, 5) IGNORE NULLS over (order by x) from (select explode(sequence(1, 3)) x)

Why are the changes needed?

Fix existing bug

Does this PR introduce any user-facing change?

No

How was this patch tested?

Modify existing unit test to cover this case

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

No

[beliefer]

The root cause is findNextRowWithNonNullInput requires assert(offset <= input.length), I think we can simplify the logic here!

    if (offset > rows.length) {
      fillDefaultValue(EmptyRow)
    } else {
      resetStates(rows)
      if (ignoreNulls) {
        findNextRowWithNonNullInput()
      } else {
        // drain the first few rows if offset is larger than zero
        while (inputIndex < offset) {
          if (inputIterator.hasNext) inputIterator.next()
          inputIndex += 1
        }
        inputIndex = offset
      }
    }

[vitaliili-db]

Thank you, I considered this but unfortunately it does not work. The actual projection (or default value) is created in doWrite so prepare needs to initialize variables such as input, inputIndex etc.

[beliefer]

I tested with the code I mentioned, It works good!

  override def prepare(rows: ExternalAppendOnlyUnsafeRowArray): Unit = {
    resetStates(rows)
    if (offset > rows.length) {
      fillDefaultValue(EmptyRow)
    } else {
      resetStates(rows)
      if (ignoreNulls) {
        findNextRowWithNonNullInput()
      } else {
        // drain the first few rows if offset is larger than zero
        while (inputIndex < offset) {
          if (inputIterator.hasNext) inputIterator.next()
          inputIndex += 1
        }
        inputIndex = offset
      }
    }
  }

[vitaliili-db]

yes, it works when ignoreNulls == true but breaks when ignoreNulls == false

In particular it will trigger last else branch in doWrite:

 else {
    (current: InternalRow) =>
      if (inputIndex >= 0 && inputIndex < input.length) {
        val r = WindowFunctionFrame.getNextOrNull(inputIterator)
        projection(r)
      } else {
        // Use default values since the offset row does not exist.
        fillDefaultValue(current)
      }
      inputIndex += 1
  }

[vitaliili-db]

Modified PR to localize change for ignoreNulls case only, please take a look

[vitaliili-db]

@beliefer please take a look.

[vitaliili-db]

@cloud-fan please take a look

[cloud-fan]

seems we can just change this to if (ignoreNulls && Math.abs(offset) > rows.length)?

[vitaliili-db]

when offset > rows.length it is safe to just set inputIndex to its value. I can add conditional inside if (ignoreNulls) if you feel strongly about it :)

[cloud-fan]

what does this do? skip everything and return default value?

[vitaliili-db]

yes, it did indirectly. To avoid confusion I modified change to limit the scope to ignoreNulls case only. Please take a look

[beliefer]

We no need Math.abs(offset) here.
findNextRowWithNonNullInput skips all the rows if offset < 0

[cloud-fan]

I think the current code is clear. It's not obvious that findNextRowWithNonNullInput works for negative offsets, and I don't think perf is a problem here.

[cloud-fan]

I think the pyspark test failure is unrelated and we can merge. @vitaliili-db how far shall we backport? is it a day-1 bug?

[vitaliili-db]

I think the pyspark test failure is unrelated and we can merge. @vitaliili-db how far shall we backport? is it a day-1 bug?

yes, it is :(. Should go as far as 11.3

[vitaliili-db]

@cloud-fan can you merge this please?

[cloud-fan]

thanks, merging to master/3.5/3.4/3/3!