Spill improvements

[sopel39]

Description

Additional context and related issues

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
(x) Release notes are required, with the following suggested text:

## General
* Improve reliability of spilling for aggregation by retaining revocable memory, preventing out‑of‑memory errors.
* Fix correctness issues when unspilling hash aggregation by using correct hash values, avoiding aggregation miscalculations.

[sopel39]

cc @osscm

[raunaqmorarka]

lgtm % comments on metrics commit, we can take that to separate PR if you want to merge the remaining first

[raunaqmorarka]

lgtm % #25892 (comment)

[pettyjamesm]

The use of flatten and create here makes this code a little difficult to follow since most of the logic only needs to be called once. Could this be refactored to a transform that runs after the spillToDisk() unblocks?

[sopel39]

I don't think it would make it much simpler. Think of flatten processor as a control-loop that delegates to other actions.

We could maybe add more WorkProcessor constructs, e.g:

WorkProcessor.of()
  .blockingOptionally(() -> {
        checkState(hasPreviousSpillCompletedSuccessfully(), "Previous spill hasn't yet finished");
        updateMemory();
        if (localRevocableMemoryContext.getBytes() > 0) {
            // No spill happened, try to build result from memory
            if (spiller.isEmpty()) {
                // No spill happened, try to build result from memory. Revocable memory needs to be converted to user memory as producing output stage is no longer revocable.
                long currentRevocableBytes = localRevocableMemoryContext.getBytes();
                localRevocableMemoryContext.setBytes(0);
                if (!localUserMemoryContext.trySetBytes(localUserMemoryContext.getBytes() + currentRevocableBytes)) {
                    // TODO: this might fail (even though we have just released memory), but we don't
                    // have a proper way to atomically convert memory reservations
                    localRevocableMemoryContext.setBytes(currentRevocableBytes);
                    // spill since revocable memory could not be converted to user memory immediately
                    return Optional.of(spillToDisk());
                }
            }
            else if (!shouldMergeWithMemory(getSizeInMemoryWhenUnspilling())) {
                return Optional.of(spillToDisk());
            }
        }
        return Optional.empty();
  })
  .flatReturn(() -> {
        checkState(hasPreviousSpillCompletedSuccessfully(), "Previous spill hasn't yet finished");
        // update memory after potential spill from the previous call to buildResult
        updateMemory();
        if (spiller.isEmpty()) {
            return hashAggregationBuilder.buildResult();
        }
        return mergeFromDiskAndMemory();
  })

it's probably easier to read, but I would skip it for now