GitHub user ggevay opened a pull request:
https://github.com/apache/flink/pull/4680
[FLINK-7629] [scala] Fix KeyedStream.aggregate for nested field expressions
## What is the purpose of the change
Make `KeyedStream.aggregate` (and `sum`, `maxBy`, etc.) able to handle
nested field expressions.
## Brief change log
The first commit fixes a bug in `RecursiveProductFieldAccessor`: the
`fieldType` field should contain the type of the innermost field. I.e., if the
field expression is `"field1.field2.field3"`, then it should be the type of
`field3`, and not `field1`.
The second commit modifies the string overload of `KeyedStream.aggregate`
to not use `fieldNames2Indices` to resolve the given field expression, but
directly create a `SumAggregator` or `ComparableAggregator`, whose constructors
call `FieldAccessorFactory.getAccessor`, which correctly resolves nested field
expressions like `"field1.field2.field3"`.
## Verifying this change
- Extended the tests in `FieldAccessorTest` to catch the `fieldType` issue.
- Added a test which calls `KeyedStream.sum` with a nested field expression.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? The feature was already
documented in the JavaDocs of the aggregation methods of `KeyedStream`. This PR
just brings the functionality up-to-date with the documentation.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ggevay/flink agg-scala-FLINK-7629
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4680.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4680
----
commit 56b524140bde7f55f5680be3a1a936a1d710722a
Author: Gabor Gevay <[email protected]>
Date: 2017-09-17T18:55:13Z
[FLINK-7629] [scala] Fix RecursiveProductFieldAccessor.fieldType
commit c605e8957de6068de6620108a643e5d8252c9d14
Author: Gabor Gevay <[email protected]>
Date: 2017-09-17T19:03:46Z
[FLINK-7629] [scala] Fix KeyedStream.aggregate for nested field expressions
----
---
