ARROW-13530: [C++] Implement cumulative sum compute function

[JabariBooker]

Creating new compute function to perform a cumulative sum on a given array/vector.

[github-actions]

https://issues.apache.org/jira/browse/ARROW-13530

[github-actions]

⚠️ Ticket has not been started in JIRA, please click 'Start Progress'.

[lidavidm]

I thought we can't implement this until we decide how to handle ordering within the query engine? CC @westonpace I may not have the context

[lidavidm]

Oh, or it can just be a vector function, which is fine, though not usable within the query engine.

[westonpace]

I think we can write the scalar function for users calling into compute directly. I agree it would be somewhat meaningless today if applied in, for example, a project expression.

Can we have it take a starting value as one of the function options?

Then we can use this whenever we get around to implementing the feature in the execution engine.

[edponce]

@JabariBooker The following are the main actions to complete when adding a compute function to C++, including Python bindings:

• Implement function/kernels in C++ .cc file
• Add function declaration to C++ public API
• Add function to C++ docs
• Only applicable if new FunctionOptions are defined:
  • Declare and define C++ XXXOptions
  • Add options name to Python imports in pyarrow/includes/compute.py
  • Cython header binding for options pyarrow/includes/libarrow.pxd
  • Cython/Python definitions for options pyarrow/_compute.pyx
  • Python test for XXXOptions
• Python tests for compute function
• Add function and FunctionOptions to Python docs

[edponce]

I think we can write the scalar function for users calling into compute directly. I agree it would be somewhat meaningless today if applied in, for example, a project expression.

@westonpace Why a ScalarFunction and not a VectorFunction? I understand the ordering issue, but...

An Array can be:

• summed up into a single total sum
• cumulative sum resulting in an Array with partial sums

[westonpace]

@edponce

@westonpace Why a ScalarFunction and not a VectorFunction? I understand the ordering issue, but...

An Array can be:

* summed up into a single total sum

This would be a vector function but we already have this with sum right?

* cumulative sum resulting in an `Array` with partial sums

Isn't this a scalar function? I may be misunderstanding the term. My current interpretation is "each row has a single output value".

[edponce]

@westonpace Oops! I forgot that sum existed.
A ScalarFunction can operate independently on each element of the Array, but a cumulative sum requires the sum of the previous element to compute the next one (or the entire Array, depends on impl.), so this pattern I think conforms more to a VectorFunction.

[westonpace]

A ScalarFunction can operate independently on each element of the Array

Given that definition I agree this is a vector function. I was thinking only of the shape of the return value and not the statefull-ness of the function.

[edponce]

...well..., I just noticed that my definition is not exactly the definition stated in the source code, so it would be great to have other opinions on this.
cc @lidavidm @pitrou

[lidavidm]

This sounds like a function "whose behavior depends on the values of the entire arrays passed", no?

[edponce]

@lidavidm I agree.
Also, a ScalarFunction should conform to this iteration pattern, and cumulative sum does not, so it is a VectorFunction.

[westonpace]

From a consumption standpoint the concept of "scalar expression" is pretty important:

Oracle: https://docs.oracle.com/cd/B19306_01/server.102/b14200/expressions010.htm
CockroachDb: https://www.cockroachlabs.com/docs/stable/scalar-expressions.html
Substrait: https://substrait.io/expressions/scalar_functions/

For example, only a scalar expression can be used in a project expression. These are contrasted with "table expressions" which can return multiple rows for each execution.

If Arrow wants to have a concept of "scalar function" and "vector function" that is different then I think that is "ok but confusing". For example, to validate a plan, we can inspect the return value of the function, and ensure that it is scalar, to decide if an expression is a "scalar expression" regardless of whether it is a "scalar function".

[lidavidm]

We already do that verification:

arrow/cpp/src/arrow/compute/exec/expression.cc

Line 283 in e989fb3

return call->function->kind() == compute::Function::SCALAR;

[westonpace]

So this particular function can be a vector function because it would be somewhat meaningless to use in a project expression anyways. Is there any example where a vector function might want to be used in a project expression? Maybe it's a moot point and we can decide "scalar expression" means "single result AND stateless"

[lidavidm]

I almost wonder if cumulative sum and other such functions should be their own (sub)class, though, since they can be processed incrementally with some state (and that state can be represented easily as an Arrow scalar), at least for the ones Pandas supports (min/max/sum/product).

[lidavidm]

And then we could have an exec node that knows how to handle them appropriately.

[westonpace]

Ok. I agree with you both. "scalar" implies stateless. This function is not stateless and thus not scalar. Whether we want to call it vector or something else we can figure out later. I suspect whatever pattern we end up needing will be useful when implementing window functions.

[lidavidm]

Is there any example where a vector function might want to be used in a project expression?

Hmm, I can imagine ones like fill_null, drop_null might be desirable. drop_null is an example of something which isn't stateful, but which is not scalar. And fill_null_forward sorta fits into a 'cumulative' function (not backward unless you want to support directionality). I guess these don't really fit as projections, but rather as postprocessing steps (it doesn't really make sense to say drop_null(a) + drop_null(b), but you could have `a + b and then drop_null)

[lidavidm]

Yeah for our purposes let's call this vector, and we can see how things to as we add more support in the query engine.

[edponce]

Thank you both for this insightful discussion.
Created ARROW-15832 to capture this discussion (still need to add more details).

[pitrou]

Thanks for the update @JabariBooker and sorry for the delay. Here are a couple more concerns.

[pitrou]

Sorry for the delay @JabariBooker . This looks great to me, and I'm going to merge now. Thank you for contributing this!

[ursabot]

Benchmark runs are scheduled for baseline = 931422b and contender = b851392. b851392 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished ⬇️0.0% ⬆️0.0%] ec2-t3-xlarge-us-east-2
[Failed ⬇️0.78% ⬆️0.47%] test-mac-arm
[Finished ⬇️0.36% ⬆️0.0%] ursa-i9-9960x
[Finished ⬇️0.79% ⬆️0.2%] ursa-thinkcentre-m75q
Buildkite builds:
[Finished] b8513920 ec2-t3-xlarge-us-east-2
[Finished] b8513920 test-mac-arm
[Finished] b8513920 ursa-i9-9960x
[Finished] b8513920 ursa-thinkcentre-m75q
[Finished] 931422bb ec2-t3-xlarge-us-east-2
[Failed] 931422bb test-mac-arm
[Finished] 931422bb ursa-i9-9960x
[Finished] 931422bb ursa-thinkcentre-m75q
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

[ursabot]

['Python', 'R'] benchmarks have high level of regressions.
ursa-i9-9960x

[wesm]

I'm refactoring all the vector kernels and I observed that the behavior of this kernel is inconsistent between scalar and array inputs:

A scalar null returns an array of length with 0 while an array with a single null returns an array with one null

In [5]: pc.cumulative_sum(pa.array([None], type='int32')[0], skip_nulls=True)
Out[5]: 
<pyarrow.lib.Int32Array object at 0x7fa80d3e7640>
[
  0
]

In [6]: pc.cumulative_sum(pa.array([None], type='int32'), skip_nulls=True)
Out[6]: 
<pyarrow.lib.Int32Array object at 0x7fa80d3e7700>
[
  null
]

I'm not sure this is right -- I can preserve this behavior, but do we want to fix it?

[wesm]

(it's also unclear to me that supporting scalar inputs to this function is useful, but that's a separate question)

[westonpace]

In the exec plan / engine (e.g. everywhere using ExecBatch), scalar columns have length (part of the reason I'm a fan of treating scalars everywhere as RLE encoded arrays is to easily express this).

For this function, if we are going to allow it to be called as a "scalar function", it must not change the length of incoming arrays. So I would expect:

A scalar null with length 0 should return an empty array or a scalar null with length 0.
A scalar null with length X should return an array of nulls with length X or a scalar null with length X.

[westonpace]

I suppose, if the kernel functions are going to keep having scalars without length then the correct thing to do would be to always output either a scalar or an array of length 1.

[wesm]

In the Acero execution engine, vector functions won't be supported in expressions aside from window functions (I guess). I'm inclined to simply disable the scalar input path here since it is ill-defined right now. I'll do that in my forthcoming patch

[wesm]

I suppose, if the kernel functions are going to keep having scalars without length then the correct thing to do would be to always output either a scalar or an array of length 1.

In ARROW-16577 (which I'm going to tackle within the next week hopefully), I'm going to remove the all-scalar input path from all kernels and up-promote ExecBatch with all scalars to be ExecSpan with arrays of length 1 (and unbox the output to be a scalar again if it's appropriate).

[westonpace]

In ARROW-16577 (which I'm going to tackle within the next week hopefully), I'm going to remove the all-scalar input path from all kernels and up-promote ExecBatch with all scalars to be ExecSpan with arrays of length 1 (and unbox the output to be a scalar again if it's appropriate).

An ExecBatch with all scalars does not necessarily have a length of 1.

[westonpace]

In the Acero execution engine, vector functions won't be supported in expressions aside from window functions (I guess). I'm inclined to simply disable the scalar input path here since it is ill-defined right now. I'll do that in my forthcoming patch

I believe there has been some desire for a "table UDF" which would require a new TableUdfNode (i.e. it would not use the project node). As far as I can tell there are no rules for a table UDF so it is a free for all. So this could qualify as a "table function".

However, table functions are not yet defined / implemented so they are in the same boat as window functions (which is probably the most correct category for this). So +1 to the idea of just removing the path for now.

[wesm]

An ExecBatch with all scalars does not necessarily have a length of 1.

I think we're talking about different things — many of the ScalarKernels have two implementations: one for all scalars which returns a scalar, and another for arrays that returns an array. I'm just talking about nixing the first code path such that the scalars -> scalar code path runs through a common implementation rather than a duplicated one.

[westonpace]

I think I see your point. Is the code path for array x scalar remaining (e.g. add(field_ref("x"), 7))?

[wesm]

Yes, definitely. I'm just referring to the implementation of e.g. f(scalar) -> scalar) or g(scalar, scalar) -> scalar — the ability to perform these operations will remain but the implementation will use the array implementation allowing us to simply delete the all-scalar implementation. See for example what it looks like to have duplicate implementations of a scalar kernel like "map_lookup"

https://github.com/apache/arrow/blob/master/cpp/src/arrow/compute/kernels/scalar_nested.cc#L614