You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by ggevay <gi...@git.apache.org> on 2015/05/18 02:02:18 UTC
[GitHub] flink pull request: [streaming] Fast calculation of medians of win...
GitHub user ggevay opened a pull request:
https://github.com/apache/flink/pull/684
[streaming] Fast calculation of medians of windows [WIP]
I started working on my Google Summer of Code Project. I decided to start with the fast calculation of medians of windows. The MedianPreReducer keeps track of the median at every window change by keeping the lower and upper halves of the window in two multisets. The updates have logarithmic runtime with the current window size. (Adding an element to or removing an element from the window involves moving at most one of the other elements between the lower and upper halves.)
The parameters of WindowedDataStream.median is the same as for the sum, min, max, etc. methods. (But here, the specified field can only be of Double type.) I implemented a few helper classes for accessing the field that is specified by these parameters. I think that the other methods with similar field selection (sum, min, max, etc.) could also be refactored to use these helper classes, so that the code duplication in SumAggregator and ComparableAggregator could be eliminated (the duplication of both the field access logic, and the logic of the byAggregate comparisons). Should I do this refactoring?
Currently, the implementation only handles non-grouped data streams, but I plan to create a GroupedMedianPreReducer (hence the [WIP] flag).
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/ggevay/flink median
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/684.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 #684
----
commit 7131289e9f7ab56ec82f12c3a25f146a17354f61
Author: Gabor Gevay <gg...@gmail.com>
Date: 2015-05-17T23:59:31Z
[streaming] Fast calculation of medians of windows
----
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-103178282
Thank you for your comments.
@rmetzger:
We will discuss creating the Jira issues with @mbalassi.
I fixed the catching of the IllegalAccessException.
The assertions seem to be enabled for the unit tests. (Surefire enables them by default, and I don't see them disabled. I also tried it by temporarily adding an 'assert false', and the test failed.) It also seems to me, that an assert is more appropriate here, as it doesn't slow the code down when we are not unit testing. And if this assert fails, that indicates some internal problem (I mean that the user shouldn't be able to make this condition not hold here).
@mbalassi:
Ok, thanks, we will discuss these points in person then.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay closed the pull request at:
https://github.com/apache/flink/pull/684
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-113907348
I merged Marton's changes (moving to contrib) to this PR.
I also added a paragraph to the streaming programming guide. I couldn't think of any better name for WindowedDataStreamStatistics.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-103035236
In general I think we should discuss whether we make Gabor's GSoC project part of the core streaming API or have them separately in a streaming windowing statistics package or even maven module. For this specific change I am personally ambivalent whether we should have this in core.
@rmetzger: Good point with JIRA. There is no ticket yet for Gabor's work, I'll discuss the issues he is addressing later this week and we will break it up to a number of issues.
@ggevay: The refactor suggestion of yours is desirable. Let us make sure to sit together in person later this week now that my trip to Budapest is confirmed.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-114083769
Concerning my comment about where to put the code:
There is so much code being added to Flink right now. The committers have to be ultimately the ones that maintain the code. That should make us think hard about what we really want to add.
What is the motivation to add this to the core projects? Is it a frequently needed functionality, from users?
This code seems to be totally okay with running "on top of Flink", as a library function. Just because it can be added is not really a reason that we can follow any more at this point.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/684#discussion_r32888221
--- Diff: flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedDataStream.scala ---
@@ -339,4 +335,30 @@ class WindowedDataStream[T](javaStream: JavaWStream[T]) {
javaStream.getDiscretizedStream.getExecutionEnvironment).scalaClean(f)
}
+ /**
+ * Gives the median of the current window at the specified field at every trigger.
+ * The type of the field can only be Double (as the median of integers might be a
+ * fractional number).
+ *
+ * The median is updated online as the window changes, and the runtime of
+ * one update is logarithmic with the current window size.
+ *
+ */
+ def median(pos: Int): WindowedDataStream[T] = {
--- End diff --
Can you also update the streaming programming guide with this new method?
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-110739389
I did the Scala part, and added a Jira (FLINK-2145).
As for the placement:
@StephanEwen:
> I think we should put the code into a streaming operator library. The streaming core is undergoing heavy changes, the windowing may be reworked in the near future, and there are plans to use managed memory for window buffer contents.
I am not sure how putting this into a separate library helps a future refactoring of windowing. Most of the median code is inside the new PreReducers, so it is probably already separated from the rest of the code as well as it can be.
(Also, if you are worried about bloating the API for the users, I don't think median would hurt in that respect either: the function of a method named "median" should be pretty clear to the users.)
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by mbalassi <gi...@git.apache.org>.
Github user mbalassi commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-113820597
Merged the first commit as that was already clarified. As for the second one the placement to either `streaming-core` or `streaming-contrib` is still unclear. I moved it to contrib on the following branch as a subclass of `WindowedDataStream`. [1] I admit that the naming is weak, any suggestions are very welcome on that.
@ggevay, @gyfora what do you think?
[1] https://github.com/mbalassi/flink/commit/956145716dca11ddaffd4b9a0e95f7f9546a055d
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-195395736
Is this still a relevant PR with all the changes to the runtime in recent months? If not, can we close this?
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-105519930
Concerning Marton's comment:
I think we should put the code into a streaming operator library. The streaming core is undergoing heavy changes, the windowing may be reworked in the near future, and there are plans to use managed memory for window buffer contents.
In light of that, I would suggest to keep these changes in a library for now. It would be easier for both parties that way.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/684#discussion_r32888195
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/FieldAccessorTest.java ---
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.util;
+
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+// This only tests a fraction of FieldAccessor. The other parts are tested indirectly by AggregationFunctionTest.
+public class FieldAccessorTest {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void arrayFieldAccessorTest() {
+ int[] a = new int[]{3,5};
+ FieldAccessor<int[], Integer> fieldAccessor =
+ (FieldAccessor<int[], Integer>) (Object)
+ FieldAccessor.create(1, PrimitiveArrayTypeInfo.getInfoFor(a.getClass()), null);
+
+ assertEquals(Integer.class, fieldAccessor.getFieldType().getTypeClass());
+
+ assertEquals((Integer)a[1], fieldAccessor.get(a));
+
+ a = fieldAccessor.set(a, 6);
+ assertEquals((Integer)a[1], fieldAccessor.get(a));
+
+
+
+ Integer[] b = new Integer[]{3,5};
+ FieldAccessor<Integer[], Integer> fieldAccessor2 =
+ (FieldAccessor<Integer[], Integer>) (Object)
+ FieldAccessor.create(1, BasicArrayTypeInfo.getInfoFor(b.getClass()), null);
+
+ assertEquals(Integer.class, fieldAccessor2.getFieldType().getTypeClass());
+
+ assertEquals((Integer)b[1], fieldAccessor2.get(b));
+
+ b = fieldAccessor2.set(b, 6);
+ assertEquals((Integer)b[1], fieldAccessor2.get(b));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void tupleFieldAccessorOutOfBoundsTest() {
+ try {
+ FieldAccessor<Tuple2<Integer, Integer>, Integer> fieldAccessor =
+ (FieldAccessor<Tuple2<Integer, Integer>, Integer>) (Object)
+ FieldAccessor.create(2, TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class),
+ null);
+ fail();
+ } catch (IndexOutOfBoundsException e) {
+ // Nothing to do here
+ }
+ }
--- End diff --
Why is there no test for the POJO field accessor?
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-195402038
This is dead. Everything has totally changed that this was based on. The fundamental problem is that this is based on the assumption of the old 0.9 windowing code that events come ordered by time. The idea was to update the median incrementally when the current window changes, but the new windowing code is building several windows at the same time instead of windows gradually changing into each other, so I don't think that this idea could be adapted to the current version.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-102964703
Hi, thank you for contributing to Flink!
Is there a JIRA issue available for this task?
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/684#discussion_r30488224
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/windowbuffer/MedianPreReducer.java ---
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.windowbuffer;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.streaming.api.windowing.StreamWindow;
+import org.apache.flink.streaming.api.windowing.helper.fieldaccessors.FieldAccessorCreator;
+import org.apache.flink.util.Collector;
+import org.apache.flink.streaming.api.windowing.helper.fieldaccessors.FieldAccessor;
+
+import java.util.TreeMap;
+import java.util.Map.Entry;
+import java.util.Deque;
+import java.util.ArrayDeque;
+import java.util.Comparator;
+import java.io.Serializable;
+
+/**
+ * Non-grouped pre-reducer for calculating median with any eviction policy.
+ */
+public class MedianPreReducer<T> extends WindowBuffer<T> implements PreAggregator, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ // This is for getting and setting the field specified in the parameters to the median call.
+ private FieldAccessor<T, Double> fieldAccessor;
+
+ private TypeSerializer<T> serializer;
+
+ // These contain lower-than-median and higher-than-median elements.
+ // Elements that compare equal to the median can be in either.
+ // The invariant is that low always contains the same amount or one more element than high.
+ // (see the assert in updateMedian)
+ TreeMultiset<T>
+ low = new TreeMultiset<T>(new CompareOnField()),
+ high = new TreeMultiset<T>(new CompareOnField());
+ Deque<T> elements = new ArrayDeque<T>();
+
+ T median;
+
+ public MedianPreReducer(int pos, TypeInformation<T> typeInfo, ExecutionConfig config) {
+ this.fieldAccessor = FieldAccessorCreator.create(pos, typeInfo);
+ this.serializer = typeInfo.createSerializer(config);
+ }
+
+ public MedianPreReducer(String field, TypeInformation<T> typeInfo, ExecutionConfig config) {
+ this.fieldAccessor = FieldAccessorCreator.create(field, typeInfo, config);
+ this.serializer = typeInfo.createSerializer(config);
+ }
+
+ private void updateMedian() {
+ assert low.size() == high.size() || low.size() == high.size() + 1;
--- End diff --
Assertions are not enabled in Flink.
I would suggest to use Guava's preconditions.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-114086691
I see your point, no problem.
Although, I would like to put my next pull request (856) to the core, because that code would be very awkward to put into contrib. But I will put later things into contrib.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on a diff in the pull request:
https://github.com/apache/flink/pull/684#discussion_r31030179
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/util/FieldAccessor.java ---
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.util;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.runtime.PojoComparator;
+
+import java.io.Serializable;
+import java.lang.reflect.Array;
+import java.util.List;
+
+// These classes encapsulate the logic of accessing a field specified by the user as either an index
--- End diff --
This should use proper JavaDoc style
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on the pull request:
https://github.com/apache/flink/pull/684#issuecomment-105249478
I coded the grouped case and did the refactoring of SumAggregator and ComparableAggregator to use the new FieldAccessor class to access the user-specified field. The first commit does the refactoring and the second adds the median calculation.
As part of the refactoring I also fixed a small bug: SimpleComparableAggregator.reduce was not handling the byAggregate case.
I also moved the logic of DataStream.getClassAtPos and checkFieldRange to FieldAccessor.
The refactoring also solves FLINK-2039. (by the first two lines of the second overload of FieldAccessor.create)
There are three things left to do: deciding where will the median be placed, creating a Jira, and possibly adding the median to the Scala API. I will do these after speaking with @mbalassi tomorrow.
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/684#discussion_r30488171
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/windowing/helper/fieldaccessors/PojoFieldAccessor.java ---
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.windowing.helper.fieldaccessors;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.runtime.PojoComparator;
+
+import java.io.Serializable;
+import java.lang.reflect.Field;
+import java.util.List;
+
+public class PojoFieldAccessor<R, F> implements FieldAccessor<R, F>, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ PojoComparator comparator;
+ Field keyField;
+
+ PojoFieldAccessor(String field, TypeInformation<R> type, ExecutionConfig config) {
+ if (!(type instanceof CompositeType<?>)) {
+ throw new IllegalArgumentException(
+ "Key expressions are only supported on POJO types and Tuples. "
+ + "A type is considered a POJO if all its fields are public, or have both getters and setters defined");
+ }
+
+ @SuppressWarnings("unchecked")
+ CompositeType<R> cType = (CompositeType<R>) type;
+
+ List<CompositeType.FlatFieldDescriptor> fieldDescriptors = cType.getFlatFields(field);
+
+ int logicalKeyPosition = fieldDescriptors.get(0).getPosition();
+ Class<?> keyClass = fieldDescriptors.get(0).getType().getTypeClass();
+
+ if (cType instanceof PojoTypeInfo) {
+ comparator = (PojoComparator<R>) cType.createComparator(
+ new int[] { logicalKeyPosition }, new boolean[] { false }, 0, config);
+ } else {
+ throw new IllegalArgumentException(
+ "Key expressions are only supported on POJO types. "
+ + "A type is considered a POJO if all its fields are public, or have both getters and setters defined");
+ }
+
+ keyField = comparator.getKeyFields()[0];
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public F get(R record) {
+ return (F) comparator.accessField(keyField, record);
+ }
+
+ @Override
+ public R set(R record, F fieldValue) {
+ try {
+ keyField.set(record, fieldValue);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException("");
--- End diff --
Add message + chain IllegalAccessException
---
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] flink pull request: [streaming] Fast calculation of medians of win...
Posted by ggevay <gi...@git.apache.org>.
Github user ggevay commented on a diff in the pull request:
https://github.com/apache/flink/pull/684#discussion_r32892426
--- Diff: flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/util/FieldAccessorTest.java ---
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.util;
+
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+// This only tests a fraction of FieldAccessor. The other parts are tested indirectly by AggregationFunctionTest.
+public class FieldAccessorTest {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void arrayFieldAccessorTest() {
+ int[] a = new int[]{3,5};
+ FieldAccessor<int[], Integer> fieldAccessor =
+ (FieldAccessor<int[], Integer>) (Object)
+ FieldAccessor.create(1, PrimitiveArrayTypeInfo.getInfoFor(a.getClass()), null);
+
+ assertEquals(Integer.class, fieldAccessor.getFieldType().getTypeClass());
+
+ assertEquals((Integer)a[1], fieldAccessor.get(a));
+
+ a = fieldAccessor.set(a, 6);
+ assertEquals((Integer)a[1], fieldAccessor.get(a));
+
+
+
+ Integer[] b = new Integer[]{3,5};
+ FieldAccessor<Integer[], Integer> fieldAccessor2 =
+ (FieldAccessor<Integer[], Integer>) (Object)
+ FieldAccessor.create(1, BasicArrayTypeInfo.getInfoFor(b.getClass()), null);
+
+ assertEquals(Integer.class, fieldAccessor2.getFieldType().getTypeClass());
+
+ assertEquals((Integer)b[1], fieldAccessor2.get(b));
+
+ b = fieldAccessor2.set(b, 6);
+ assertEquals((Integer)b[1], fieldAccessor2.get(b));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void tupleFieldAccessorOutOfBoundsTest() {
+ try {
+ FieldAccessor<Tuple2<Integer, Integer>, Integer> fieldAccessor =
+ (FieldAccessor<Tuple2<Integer, Integer>, Integer>) (Object)
+ FieldAccessor.create(2, TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class),
+ null);
+ fail();
+ } catch (IndexOutOfBoundsException e) {
+ // Nothing to do here
+ }
+ }
--- End diff --
It is tested by AggregationFunctionTest.
ProductFieldAccessor is tested by TopSpeedWindowingExampleITCase (or it will be, after fixing FLINK-2255).
I will add this as a comment to FieldAccessorTest.java.
---
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.
---