[
https://issues.apache.org/jira/browse/FLINK-7629?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16169412#comment-16169412
]
ASF GitHub Bot commented on FLINK-7629:
---------------------------------------
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
----
> Scala stream aggregations should support nested field expressions
> -----------------------------------------------------------------
>
> Key: FLINK-7629
> URL: https://issues.apache.org/jira/browse/FLINK-7629
> Project: Flink
> Issue Type: Bug
> Components: Scala API, Streaming
> Reporter: Gabor Gevay
> Assignee: Gabor Gevay
> Priority: Minor
> Fix For: 1.4.0
>
>
> In the Scala API, {{KeyedStream.maxBy}} and similar methods currently only
> work with a field name, and not with nested field expressions, such as
> "fieldA.fieldX". (Their documentation says this should work.)
> The reason for this is that the string overload of {{KeyedStream.aggregate}}
> uses {{fieldNames2Indices}} and then calls the integer overload. Instead, it
> should create a {{SumAggregator}} or {{ComparableAggregator}} directly, as
> the integer overload does (and as the Java API does). The ctors of
> {{SumAggregator}} or {{ComparableAggregator}} will call
> {{FieldAccessorFactory.getAccessor}}, which will correctly handle a nested
> field expression.
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
