[Data] Fixing aggregation protocol to be appropriately associative

[alexeykudinkin]

Why are these changes needed?

This is a follow-up for #50585, addressing an issue of its combination sequence not being appropriately associative.

Primary hurdle for implementing properly associative aggregation in the presence null values is to be able to distinguish between:

• Empty accumulator
• Accumulator holding single value that is null

To achieve that in the presence of null values following semantic is established.

Case of ignore_nulls=True:

• Combination protocol

  • If current accumulator is null (ie empty), return new accumulator
  • If new accumulator is null (ie empty), return cur
  • Otherwise combine (current and new)

• Identity (zero) value is None (ie simulating 'empty' sequence)

Case of ignore_nulls=False:

• Combination protocol

  • If new accumulator is null (ie has null in the sequence, b/c we're
    NOT ignoring nulls), return it
  • If current accumulator is null (ie had null in the prior sequence,
    b/c we're NOT ignoring nulls), return it
  • Otherwise combine (current and new)

• Identity (zero) value is an actual zero value for the operation (0 for count, sum, -inf for max, +inf for min, etc)

Changes

• Revisited aggregation protocol rebasing it onto _Optional
• Added explicit aggregation protocol test
• Fixed Quantile finalization seq
• Fixed Std/Mean incorrectly handling NaNs

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[raulchen]

nit, we can move ignore_null check out of this function. So it doesn't need to be checked for each single record

[alexeykudinkin]

Good call! Fixed for combination seq as well

[raulchen]

Need add the following at the end of this file.

if __name__ == "__main__":
    sys.exit(pytest.main(["-v", __file__]))

[alexeykudinkin]

Ah, keep forgetting about it

[raulchen]

Can you also add some unit test for these wrappers?

[alexeykudinkin]

These are all tested explicitly in test_null_safe_aggregation_protocol