You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "jiangxin369 (via GitHub)" <gi...@apache.org> on 2023/02/06 02:13:13 UTC
[GitHub] [flink-ml] jiangxin369 opened a new pull request, #203: [FLINK-30688] Disable Kryo fallback in the remain Flink ML tests
jiangxin369 opened a new pull request, #203:
URL: https://github.com/apache/flink-ml/pull/203
<!--
*Thank you very much for contributing to Apache Flink ML - we are happy that you want to help us improve Flink ML. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to one [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] Title of the pull request`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Disable Kryo fallback in Flink ML tests to ensure that all Flink ML algorithms are not using Kryo.
## Brief change log
*(for example:)*
- Configures StreamExecutionEnvironment using env.getConfig().disableGenericTypes() in all Flink ML tests.
- Add TypeInfoFactory for QuantileSummary
- Add TypeInfoFactory for Vector
## 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)
## Documentation
- Does this pull request introduce a new feature? (no)
- If yes, how is the feature documented? (not applicable)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 merged pull request #203: [FLINK-30688][followup] Disable Kryo fallback for tests in flink-ml-lib
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 merged PR #203:
URL: https://github.com/apache/flink-ml/pull/203
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1097134531
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/quantile/QuantileSummary.java:
##########
@@ -395,12 +400,14 @@ public QueryResult(int index, long minRankAtIndex, double percentile) {
* <li>delta: the maximum span of the rank.
* </ul>
*/
- private static class StatsTuple implements Serializable {
+ public static class StatsTuple implements Serializable {
private static final long serialVersionUID = 1L;
- private final double value;
+ private double value;
private long g;
private long delta;
+ public StatsTuple() {}
Review Comment:
Is this change needed to make this class POJO? If yes, then why its member variable is private?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/quantile/QuantileSummary.java:
##########
@@ -36,6 +37,7 @@
* algorithm proposed in the paper: "Space-efficient Online Computation of Quantile Summaries" by
* Greenwald, Michael and Khanna, Sanjeev. (https://doi.org/10.1145/375663.375670)
*/
+@TypeInfo(QuantileSummaryTypeInfoFactory.class)
Review Comment:
Would it be simpler to put both `QuantileSummary` and `QuantileSummaryTypeInfoFactory` directly under the package
`common/util`?
Note that `QuantileSummaryTypeInfoFactory` should be an internal class that users should not need to access directly.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/lsh/LSHModel.java:
##########
@@ -373,34 +395,45 @@ public TopKFunction(String distCol, int numNearestNeighbors) {
}
@Override
- public PriorityQueue<Row> createAccumulator() {
- return new PriorityQueue<>(numNearestNeighbors, new DistColComparator(distCol));
+ public List<Row> createAccumulator() {
+ return new ArrayList<>(numNearestNeighbors);
}
@Override
- public PriorityQueue<Row> add(Row value, PriorityQueue<Row> accumulator) {
- if (accumulator.size() == numNearestNeighbors) {
- Row peek = accumulator.peek();
- if (accumulator.comparator().compare(value, peek) < 0) {
- accumulator.poll();
- }
+ public List<Row> add(Row value, List<Row> accumulator) {
+ if (topKRows == null) {
+ topKRows = new PriorityQueue<>(numNearestNeighbors, new DistColComparator(distCol));
+ topKRows.addAll(accumulator);
}
- accumulator.add(value);
- return accumulator;
+ insert(value, topKRows);
+ return new ArrayList<>(topKRows);
Review Comment:
It would be pretty inefficient to copy the entire list every time we add a value.
How about we use `PriorityQueue<Row>` as the accumulator and let `getResult(List<Row> accumulator)` return `List<ROW>`? I think we will only need to serialize the output of `getResult` but not the accumulator itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1100964409
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/QuantileSummary.java:
##########
@@ -367,6 +373,26 @@ public double getRelativeError() {
return relativeError;
}
+ public int getCompressThreshold() {
Review Comment:
No, it is for unittests currently. But I didn't annotate it with `VisibleForTesting` because I think it is reasonable for users to read the information of a `QuantileSummary`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1097419512
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/quantile/QuantileSummary.java:
##########
@@ -395,12 +400,14 @@ public QueryResult(int index, long minRankAtIndex, double percentile) {
* <li>delta: the maximum span of the rank.
* </ul>
*/
- private static class StatsTuple implements Serializable {
+ public static class StatsTuple implements Serializable {
private static final long serialVersionUID = 1L;
- private final double value;
+ private double value;
private long g;
private long delta;
+ public StatsTuple() {}
Review Comment:
No, it is not. It's needed to make the class can be used by `Types.POJO`. In this way, we need to provide the field information of the class so that Flink can access these fields by reflection and does not require all member variables to be public.
But you remind me that actually, the `StatsTuple` is a standard POJO, I don't need to define the TypeInfo by `Types.POJO`, making its member variables public is a simpler way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1097419512
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/quantile/QuantileSummary.java:
##########
@@ -395,12 +400,14 @@ public QueryResult(int index, long minRankAtIndex, double percentile) {
* <li>delta: the maximum span of the rank.
* </ul>
*/
- private static class StatsTuple implements Serializable {
+ public static class StatsTuple implements Serializable {
private static final long serialVersionUID = 1L;
- private final double value;
+ private double value;
private long g;
private long delta;
+ public StatsTuple() {}
Review Comment:
No, it isn't. It's just needed to make the class can be used by `Types.POJO`. In this way, we define the fields of the class so that Flink can access these fields by reflection and does not require all member variables to be public.
But you remind me that actually, the `StatsTuple` is a standard POJO, I don't need to define the TypeInfo by `Types.POJO`, making its member variables public is a simpler way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1097419512
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/quantile/QuantileSummary.java:
##########
@@ -395,12 +400,14 @@ public QueryResult(int index, long minRankAtIndex, double percentile) {
* <li>delta: the maximum span of the rank.
* </ul>
*/
- private static class StatsTuple implements Serializable {
+ public static class StatsTuple implements Serializable {
private static final long serialVersionUID = 1L;
- private final double value;
+ private double value;
private long g;
private long delta;
+ public StatsTuple() {}
Review Comment:
No, it isn't. We don't need to make its member variable public because I define the `TypeInfomation` for `StatsTuple` with `Types.POJO`. We need to
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1100963438
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/typeinfo/QuantileSummaryTypeInfoFactory.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.ml.common.typeinfo;
+
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.ml.common.util.QuantileSummary;
+
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Used by {@link TypeExtractor} to create a {@link TypeInformation} for implementations of {@link
+ * QuantileSummary}.
+ */
+public class QuantileSummaryTypeInfoFactory extends TypeInfoFactory<QuantileSummary> {
+
+ private static final Map<String, TypeInformation<?>> fields;
+
+ static {
+ fields = new HashMap<>();
+ fields.put("relativeError", Types.DOUBLE);
+ fields.put("compressThreshold", Types.INT);
+ fields.put("count", Types.LONG);
+ fields.put("sampled", Types.LIST(TypeInformation.of(QuantileSummary.StatsTuple.class)));
+ fields.put("headBuffer", Types.LIST(Types.DOUBLE));
+ fields.put("compressed", Types.BOOLEAN);
+ }
+
+ private static final TypeInformation<QuantileSummary> TYPE_INFO =
+ Types.POJO(QuantileSummary.class, fields);
Review Comment:
According to [1], a POJO must not contain fields that are not supported by registered serializers, while `QuantileSummary` has fields like `List<Double> headBuffer`, and `List` has no registered serializer and is treated as a generic type in Flink. So we still have to register the type info with `Types.POJO` and explicitly specify the fields.
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/serialization/types_serialization/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1097198478
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/lsh/LSHModel.java:
##########
@@ -373,34 +395,45 @@ public TopKFunction(String distCol, int numNearestNeighbors) {
}
@Override
- public PriorityQueue<Row> createAccumulator() {
- return new PriorityQueue<>(numNearestNeighbors, new DistColComparator(distCol));
+ public List<Row> createAccumulator() {
+ return new ArrayList<>(numNearestNeighbors);
}
@Override
- public PriorityQueue<Row> add(Row value, PriorityQueue<Row> accumulator) {
- if (accumulator.size() == numNearestNeighbors) {
- Row peek = accumulator.peek();
- if (accumulator.comparator().compare(value, peek) < 0) {
- accumulator.poll();
- }
+ public List<Row> add(Row value, List<Row> accumulator) {
+ if (topKRows == null) {
+ topKRows = new PriorityQueue<>(numNearestNeighbors, new DistColComparator(distCol));
+ topKRows.addAll(accumulator);
}
- accumulator.add(value);
- return accumulator;
+ insert(value, topKRows);
+ return new ArrayList<>(topKRows);
Review Comment:
Note that if the aggregator needs to be serializable by either `kryo` or our custom serializer, we can define a specific container class that has the `PriorityQueue<ROW>` as member variable and define a custom serializer for it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1100964409
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/QuantileSummary.java:
##########
@@ -367,6 +373,26 @@ public double getRelativeError() {
return relativeError;
}
+ public int getCompressThreshold() {
Review Comment:
We are not making this class POJO as the description in the below conversation.
The get method is for unittests currently. But I didn't annotate it with `VisibleForTesting` because I think it is reasonable for users to read the information of a `QuantileSummary`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on PR #203:
URL: https://github.com/apache/flink-ml/pull/203#issuecomment-1423667780
Thanks for the update. LGTM.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1097419512
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/quantile/QuantileSummary.java:
##########
@@ -395,12 +400,14 @@ public QueryResult(int index, long minRankAtIndex, double percentile) {
* <li>delta: the maximum span of the rank.
* </ul>
*/
- private static class StatsTuple implements Serializable {
+ public static class StatsTuple implements Serializable {
private static final long serialVersionUID = 1L;
- private final double value;
+ private double value;
private long g;
private long delta;
+ public StatsTuple() {}
Review Comment:
No, it is not. It's also required to make the class can be used by `Types.POJO`. In this way, we need to provide the field information of the class so that Flink can access these fields by reflection and does not require all member variables to be public.
But you remind me that actually, the `StatsTuple` is a standard POJO, I don't need to define the TypeInfo by `Types.POJO`, making its member variables public is a simpler way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #203: [FLINK-30688][followup] Disable Kryo fallback in Flink ML tests
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #203:
URL: https://github.com/apache/flink-ml/pull/203#discussion_r1099814163
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/typeinfo/PriorityQueueSerializer.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.ml.common.typeinfo;
+
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.PriorityQueue;
+
+/**
+ * TypeSerializer for {@link java.util.PriorityQueue}.
+ *
+ * @param <T> The type of inner objects.
+ */
+public class PriorityQueueSerializer<T> extends TypeSerializer<PriorityQueue<T>> {
+
+ private final Comparator<? super T> comparator;
+
+ private final TypeSerializer<T> innerSerializer;
Review Comment:
nits: can we rename it to `elementSerializer` for consistency with `LinkedListSerializer#elementSerializer` and `ListSerializer#elementSerializer`?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/typeinfo/PriorityQueueSerializer.java:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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.ml.common.typeinfo;
+
+import org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerSnapshot;
+import org.apache.flink.api.common.typeutils.base.ListSerializer;
+import org.apache.flink.api.java.typeutils.runtime.DataInputViewStream;
+import org.apache.flink.api.java.typeutils.runtime.DataOutputViewStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.util.InstantiationUtil;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Objects;
+import java.util.PriorityQueue;
+
+/**
+ * TypeSerializer for {@link java.util.PriorityQueue}.
+ *
+ * @param <T> The type of inner objects.
+ */
+public class PriorityQueueSerializer<T> extends TypeSerializer<PriorityQueue<T>> {
+
+ private final Comparator<? super T> comparator;
+
+ private final TypeSerializer<T> innerSerializer;
+
+ public PriorityQueueSerializer(
+ Comparator<? super T> comparator, TypeSerializer<T> innerSerializer) {
+ this.comparator = comparator;
+ this.innerSerializer = innerSerializer;
+ }
+
+ @Override
+ public boolean isImmutableType() {
+ return false;
+ }
+
+ @Override
+ public TypeSerializer<PriorityQueue<T>> duplicate() {
+ return new PriorityQueueSerializer<>(comparator, innerSerializer.duplicate());
+ }
+
+ @Override
+ public PriorityQueue<T> createInstance() {
+ return new PriorityQueue<>(comparator);
+ }
+
+ @Override
+ public PriorityQueue<T> copy(PriorityQueue<T> from) {
+ return new PriorityQueue<>(from);
+ }
+
+ @Override
+ public PriorityQueue<T> copy(PriorityQueue<T> from, PriorityQueue<T> reuse) {
+ return new PriorityQueue<>(from);
+ }
+
+ @Override
+ public int getLength() {
+ return -1;
+ }
+
+ @Override
+ public void serialize(PriorityQueue<T> queue, DataOutputView target) throws IOException {
+ List<T> tmpList = new ArrayList<>(queue);
+ ListSerializer<T> listSerializer = new ListSerializer<>(innerSerializer);
+ listSerializer.serialize(tmpList, target);
+ }
+
+ @Override
+ public PriorityQueue<T> deserialize(DataInputView source) throws IOException {
+ ListSerializer<T> listSerializer = new ListSerializer<>(innerSerializer);
+ List<T> tmpList = listSerializer.deserialize(source);
+ PriorityQueue<T> queue = new PriorityQueue<>(comparator);
+ queue.addAll(tmpList);
+ return queue;
+ }
+
+ @Override
+ public PriorityQueue<T> deserialize(PriorityQueue<T> reuse, DataInputView source)
+ throws IOException {
+ return deserialize(source);
+ }
+
+ @Override
+ public void copy(DataInputView source, DataOutputView target) throws IOException {
+ PriorityQueue<T> queue = deserialize(source);
Review Comment:
Would it be simpler to do the following without having to convert from/to PriorityQueue?
```
ListSerializer<T> listSerializer = new ListSerializer<>(innerSerializer);
listSerializer.copy(source, target);
```
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/util/QuantileSummary.java:
##########
@@ -367,6 +373,26 @@ public double getRelativeError() {
return relativeError;
}
+ public int getCompressThreshold() {
Review Comment:
Is this needed to make this class POJO? Would it be simpler to still keep these member variables `public`?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/typeinfo/QuantileSummaryTypeInfoFactory.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * 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.ml.common.typeinfo;
+
+import org.apache.flink.api.common.typeinfo.TypeInfoFactory;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.ml.common.util.QuantileSummary;
+
+import java.lang.reflect.Type;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Used by {@link TypeExtractor} to create a {@link TypeInformation} for implementations of {@link
+ * QuantileSummary}.
+ */
+public class QuantileSummaryTypeInfoFactory extends TypeInfoFactory<QuantileSummary> {
+
+ private static final Map<String, TypeInformation<?>> fields;
+
+ static {
+ fields = new HashMap<>();
+ fields.put("relativeError", Types.DOUBLE);
+ fields.put("compressThreshold", Types.INT);
+ fields.put("count", Types.LONG);
+ fields.put("sampled", Types.LIST(TypeInformation.of(QuantileSummary.StatsTuple.class)));
+ fields.put("headBuffer", Types.LIST(Types.DOUBLE));
+ fields.put("compressed", Types.BOOLEAN);
+ }
+
+ private static final TypeInformation<QuantileSummary> TYPE_INFO =
+ Types.POJO(QuantileSummary.class, fields);
Review Comment:
Can we just use `TypeInformation.of(QuantileSummary.class)` here after making `QuantileSummary` POJO, instead of having to explicitly specifying the `fields` above?
And do we still need `QuantileSummaryTypeInfoFactory` if `QuantileSummary` is POJO?
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/typeinfo/PriorityQueueTypeInfo.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.ml.common.typeinfo;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+import java.util.Comparator;
+import java.util.Objects;
+import java.util.PriorityQueue;
+
+/**
+ * TypeInformation for {@link java.util.PriorityQueue}.
+ *
+ * @param <T> The type of inner objects.
+ */
+public class PriorityQueueTypeInfo<T> extends TypeInformation<PriorityQueue<T>> {
+
+ private final Comparator<? super T> comparator;
+
+ private final TypeInformation<T> innerTypeInfo;
Review Comment:
nits: can we rename it to `elementTypeInfo` for consistency with `ListTypeInfo#elementTypeInfo`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org