You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by aljoscha <gi...@git.apache.org> on 2014/12/11 13:09:11 UTC
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
GitHub user aljoscha opened a pull request:
https://github.com/apache/incubator-flink/pull/263
[scala] Change ScalaAggregateOperator to use TypeSerializer
Before it was using TypeSerializerFactory. This led to problems with the
user code class loader and is not necessary.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/aljoscha/incubator-flink scala-fix-aggregate
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-flink/pull/263.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 #263
----
commit 2813da011a86e191fbf85ac25502ba4d88362295
Author: Aljoscha Krettek <al...@gmail.com>
Date: 2014-12-11T12:07:06Z
[scala] Change ScalaAggregateOperator to use TypeSerializer
Before it was using TypeSerializerFactory. This led to problems with the
user code class loader and is not necessary.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/263#discussion_r21730455
--- Diff: flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java ---
@@ -164,18 +160,8 @@ public ScalaAggregateOperator(Grouping<IN> input, Aggregations function, int fie
}
genName.setLength(genName.length()-1);
- TypeSerializer<IN> serializer = getInputType().createSerializer();
- TypeSerializerFactory<IN> serializerFactory;
- if (serializer.isStateful()) {
- serializerFactory = new RuntimeStatefulSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- } else {
- serializerFactory = new RuntimeStatelessSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- }
-
@SuppressWarnings("rawtypes")
- RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(serializerFactory, aggFunctions, fields);
+ RichGroupReduceFunction<IN, IN> function = new AggregatingUdf((TupleSerializerBase) getInputType().createSerializer(), aggFunctions, fields);
--- End diff --
Ok, will do.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on the pull request:
https://github.com/apache/incubator-flink/pull/263#issuecomment-66750006
I moved the cast and added a Validate.* check.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by hsaputra <gi...@git.apache.org>.
Github user hsaputra commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/263#discussion_r21689646
--- Diff: flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java ---
@@ -164,18 +160,8 @@ public ScalaAggregateOperator(Grouping<IN> input, Aggregations function, int fie
}
genName.setLength(genName.length()-1);
- TypeSerializer<IN> serializer = getInputType().createSerializer();
- TypeSerializerFactory<IN> serializerFactory;
- if (serializer.isStateful()) {
- serializerFactory = new RuntimeStatefulSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- } else {
- serializerFactory = new RuntimeStatelessSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- }
-
@SuppressWarnings("rawtypes")
- RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(serializerFactory, aggFunctions, fields);
+ RichGroupReduceFunction<IN, IN> function = new AggregatingUdf((TupleSerializerBase) getInputType().createSerializer(), aggFunctions, fields);
--- End diff --
Could we do explicit conversion from TypeSerializer to TupleSerializerBase instead of hard casting?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/incubator-flink/pull/263#issuecomment-66916008
Looks good, will merge...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by hsaputra <gi...@git.apache.org>.
Github user hsaputra commented on the pull request:
https://github.com/apache/incubator-flink/pull/263#issuecomment-66812856
+1
LGTM
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/263#discussion_r21689957
--- Diff: flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java ---
@@ -164,18 +160,8 @@ public ScalaAggregateOperator(Grouping<IN> input, Aggregations function, int fie
}
genName.setLength(genName.length()-1);
- TypeSerializer<IN> serializer = getInputType().createSerializer();
- TypeSerializerFactory<IN> serializerFactory;
- if (serializer.isStateful()) {
- serializerFactory = new RuntimeStatefulSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- } else {
- serializerFactory = new RuntimeStatelessSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- }
-
@SuppressWarnings("rawtypes")
- RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(serializerFactory, aggFunctions, fields);
+ RichGroupReduceFunction<IN, IN> function = new AggregatingUdf((TupleSerializerBase) getInputType().createSerializer(), aggFunctions, fields);
--- End diff --
What do you mean by explicit conversion?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by hsaputra <gi...@git.apache.org>.
Github user hsaputra commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/263#discussion_r21694186
--- Diff: flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java ---
@@ -164,18 +160,8 @@ public ScalaAggregateOperator(Grouping<IN> input, Aggregations function, int fie
}
genName.setLength(genName.length()-1);
- TypeSerializer<IN> serializer = getInputType().createSerializer();
- TypeSerializerFactory<IN> serializerFactory;
- if (serializer.isStateful()) {
- serializerFactory = new RuntimeStatefulSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- } else {
- serializerFactory = new RuntimeStatelessSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- }
-
@SuppressWarnings("rawtypes")
- RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(serializerFactory, aggFunctions, fields);
+ RichGroupReduceFunction<IN, IN> function = new AggregatingUdf((TupleSerializerBase) getInputType().createSerializer(), aggFunctions, fields);
--- End diff --
At least for now, could we change the signature of the constructor to TypeSerializer and add validate for the argument if it could be cast to TupleSerializerBase?
So add check:
Validate.isTrue(serializer instanceof TupleSerializerBase);
Casting without proper check makes me nervous.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by hsaputra <gi...@git.apache.org>.
Github user hsaputra commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/263#discussion_r21690247
--- Diff: flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java ---
@@ -164,18 +160,8 @@ public ScalaAggregateOperator(Grouping<IN> input, Aggregations function, int fie
}
genName.setLength(genName.length()-1);
- TypeSerializer<IN> serializer = getInputType().createSerializer();
- TypeSerializerFactory<IN> serializerFactory;
- if (serializer.isStateful()) {
- serializerFactory = new RuntimeStatefulSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- } else {
- serializerFactory = new RuntimeStatelessSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- }
-
@SuppressWarnings("rawtypes")
- RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(serializerFactory, aggFunctions, fields);
+ RichGroupReduceFunction<IN, IN> function = new AggregatingUdf((TupleSerializerBase) getInputType().createSerializer(), aggFunctions, fields);
--- End diff --
Ah sorry, I meant getIntputType call returns TypeSerializer and the AggregatingUdf constructor is taking TupleSerializerBase so could the constructor parameter takes TypeSerializer instead?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by aljoscha <gi...@git.apache.org>.
Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/263#discussion_r21690839
--- Diff: flink-scala/src/main/java/org/apache/flink/api/scala/operators/ScalaAggregateOperator.java ---
@@ -164,18 +160,8 @@ public ScalaAggregateOperator(Grouping<IN> input, Aggregations function, int fie
}
genName.setLength(genName.length()-1);
- TypeSerializer<IN> serializer = getInputType().createSerializer();
- TypeSerializerFactory<IN> serializerFactory;
- if (serializer.isStateful()) {
- serializerFactory = new RuntimeStatefulSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- } else {
- serializerFactory = new RuntimeStatelessSerializerFactory<IN>(
- serializer, getInputType().getTypeClass());
- }
-
@SuppressWarnings("rawtypes")
- RichGroupReduceFunction<IN, IN> function = new AggregatingUdf(serializerFactory, aggFunctions, fields);
+ RichGroupReduceFunction<IN, IN> function = new AggregatingUdf((TupleSerializerBase) getInputType().createSerializer(), aggFunctions, fields);
--- End diff --
We could maybe. If we enhanced TypeSerializer to have a method createInstance(...) that can create an instance with the given field values. This is only possible for a small subset of the types that we support: The Java Tuple Types, since we control their constructors. And Scala Case Classes, since we assume that the fields are exactly the constructor arguments, as is customary for Case Classes.
We would then need to add createInstance(...) methods to all other TypeSerializers that throw an Exception telling the user that instance creation with field values is not supported.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: [scala] Change ScalaAggregateOperato...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/incubator-flink/pull/263
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---