[GitHub] [flink] flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-534879961 ## CI report: * 8fffbbc4bc605ccedad7412ac2292d3488261fa3 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129046428) * 120f12f13987874b1738da7db1aa370f4335830a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129105662) * 8d3ff9abe90e8e23b2bd1ec12cd99dc7444f6a8d : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129205704) * 2c449c28158d24a2a65c9b01f1636d8b544d29ee : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129214214) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] ashangit commented on issue #8760: [FLINK-12869] Add yarn acls capability to flink containers
ashangit commented on issue #8760: [FLINK-12869] Add yarn acls capability to flink containers URL: https://github.com/apache/flink/pull/8760#issuecomment-535350073 @GJL will look at how I can add some test this week or the next 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938298#comment-16938298
]
Jingsong Lee commented on FLINK-13740:
--
It depends on implementation details... BinaryFormat only provide getters,
equals and hashCode from java Object, No matter how we implement it, we should
have these methods. Whether they are materialized or not should be left to the
caller to decide.
Actually, these fields are not final, they need to be controlled outside.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi…
flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi… URL: https://github.com/apache/flink/pull/9773#issuecomment-535013943 ## CI report: * a7a643c54c912e9a42b1dbf887f14eba2ac80f82 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129101985) * c0f46541c04433aaef6301b81da9b34cc1d304a8 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129208037) * b51f00a2a5eb3ff0c61caafba786e53211db8f69 : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-534879961 ## CI report: * 8fffbbc4bc605ccedad7412ac2292d3488261fa3 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129046428) * 120f12f13987874b1738da7db1aa370f4335830a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129105662) * 8d3ff9abe90e8e23b2bd1ec12cd99dc7444f6a8d : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129205704) * 2c449c28158d24a2a65c9b01f1636d8b544d29ee : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-534087366 ## CI report: * dd50b7a2af07a87280b2e358d31d0ad9345ad8c4 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743957) * abaf8ebcbd17e79ae92f9e66ed96772262919370 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129035121) * ee70e3344eda77f3d180b63460444f8e3064b2cb : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129151429) * 07a3ed3525cd48ee1e8c9d2349c48d0031ca167e : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/129206839) * 541187ec0f55a5dd3b0e390320a45bb30755caa9 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129209308) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938289#comment-16938289
]
Jark Wu commented on FLINK-13740:
-
If it is still lazy, then in my understanding, BinaryGeneric should override
some methods of BinaryFormat (getOffset, getSizeInBytes, getSegments, hashCode)
to materialize first. But these methods (getOffset, getSizeInBytes,
getSegments, hashCode) don't have a serializer parameter.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] Mrart edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb …
Mrart edited a comment on issue #9542: [FLINK-13873][metrics] Change the column family as tags for influxdb … URL: https://github.com/apache/flink/pull/9542#issuecomment-534871400 @zentol Could you help review again，I have add A config option。I am sorry for delay。 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9688: [FLINK-13056][runtime] Introduce FastRestartPipelinedRegionStrategy
flinkbot edited a comment on issue #9688: [FLINK-13056][runtime] Introduce FastRestartPipelinedRegionStrategy URL: https://github.com/apache/flink/pull/9688#issuecomment-531737820 ## CI report: * 23278b65f27154ae066841334f0d0b06885e713a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/127786107) * 5831d972c3f2a68398438015352e5a5e0a8de9da : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129102026) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi…
flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi… URL: https://github.com/apache/flink/pull/9773#issuecomment-535013943 ## CI report: * a7a643c54c912e9a42b1dbf887f14eba2ac80f82 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129101985) * c0f46541c04433aaef6301b81da9b34cc1d304a8 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129208037) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] carp84 commented on a change in pull request #9501: [FLINK-12697] [State Backends] Support on-disk state storage for spill-able heap backend
carp84 commented on a change in pull request #9501: [FLINK-12697] [State
Backends] Support on-disk state storage for spill-able heap backend
URL: https://github.com/apache/flink/pull/9501#discussion_r328435002
##
File path:
flink-state-backends/flink-statebackend-heap-spillable/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteSkipListStateMap.java
##
@@ -0,0 +1,1527 @@
+/*
+ * 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.runtime.state.heap;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.ByteBufferUtils;
+import org.apache.flink.runtime.state.StateEntry;
+import org.apache.flink.runtime.state.StateTransformationFunction;
+import org.apache.flink.runtime.state.heap.space.Allocator;
+import org.apache.flink.runtime.state.heap.space.Chunk;
+import org.apache.flink.runtime.state.heap.space.SpaceUtils;
+import org.apache.flink.runtime.state.internal.InternalKvState;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.ResourceGuard;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Set;
+import java.util.Spliterators;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+import static org.apache.flink.runtime.state.heap.SkipListUtils.HEAD_NODE;
+import static org.apache.flink.runtime.state.heap.SkipListUtils.NIL_NODE;
+import static
org.apache.flink.runtime.state.heap.SkipListUtils.NIL_VALUE_POINTER;
+
+/**
+ * Implementation of state map which is based on skip list with copy-on-write
support. states will
+ * be serialized to bytes and stored in the space allocated with the given
allocator.
+ *
+ * @param type of key
+ * @param type of namespace
+ * @param type of state
+ */
+public class CopyOnWriteSkipListStateMap extends StateMap
implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(CopyOnWriteSkipListStateMap.class);
+
+ /**
+* Default max number of logically-removed keys to delete one time.
+*/
+ private static final int DEFAULT_MAX_KEYS_TO_DELETE_ONE_TIME = 3;
+
+ /**
+* Default ratio of the logically-removed keys to trigger deletion when
snapshot.
+*/
+ private static final float DEFAULT_LOGICAL_REMOVED_KEYS_RATIO = 0.2f;
+
+ /**
+* The serializer used to serialize the key and namespace to bytes
stored in skip list.
+*/
+ private final SkipListKeySerializer skipListKeySerializer;
+
+ /**
+* The serializer used to serialize the state to bytes stored in skip
list.
+*/
+ private final SkipListValueSerializer skipListValueSerializer;
+
+ /**
+* Space allocator.
+*/
+ private final Allocator spaceAllocator;
+
+ /**
+* The level index header.
+*/
+ private final LevelIndexHeader levelIndexHeader;
+
+ /**
+* Seed to generate random index level.
+*/
+ private int randomSeed;
+
+ /**
+* The current version of this map. Used for copy-on-write mechanics.
+*/
+ private int stateMapVersion;
+
+ /**
+* The highest version of this map that is still required by any
unreleased snapshot.
+*/
+ private int highestRequiredSnapshotVersion;
+
+ /**
+* Snapshots no more than this version must have been finished, but
there may be some
+* snapshots more than this version are still running.
+*/
+ private volatile int highestFinishedSnapshotVersion;
+
+ /**
+* Maintains an ordered set of version ids that are still
[GitHub] [flink] xuyang1706 commented on a change in pull request #9184: [FLINK-13339][ml] Add an implementation of pipeline's api
xuyang1706 commented on a change in pull request #9184: [FLINK-13339][ml] Add
an implementation of pipeline's api
URL: https://github.com/apache/flink/pull/9184#discussion_r328434865
##
File path:
flink-ml-parent/flink-ml-lib/src/main/java/org/apache/flink/ml/common/MLEnvironment.java
##
@@ -0,0 +1,178 @@
+/*
+ * 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;
+
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.java.BatchTableEnvironment;
+import org.apache.flink.table.api.java.StreamTableEnvironment;
+import org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl;
+import org.apache.flink.types.Row;
+
+/**
+ * MLEnvironment hold the execution environment.
+ *
+ * @see ExecutionEnvironment
+ * @see StreamExecutionEnvironment
+ * @see BatchTableEnvironment
+ * @see StreamTableEnvironment
+ */
+public class MLEnvironment {
+ private ExecutionEnvironment env;
Review comment:
Yes, it is intended to be this for simplicity.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] carp84 commented on issue #9255: [FLINK-13034] Introduce isEmpty method for MapState
carp84 commented on issue #9255: [FLINK-13034] Introduce isEmpty method for MapState URL: https://github.com/apache/flink/pull/9255#issuecomment-535334308 > What's more, since this adds a method in `MapState` interface which is `PublicEvolving`, we will need a release note. And don't forget about the release note. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
flinkbot edited a comment on issue #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0 URL: https://github.com/apache/flink/pull/9712#issuecomment-532984287 ## CI report: * f2b7710f65f478342de389c8e099799287ddf3f9 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128285030) * 19872fd562f606e9738f5b5a77f7a1d3e93a39b9 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128302786) * 211f984c5380d61cf78f94e8ddcb4576f5c3f751 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128442800) * ed7ead402b73f1d5958f13c050dcda4b22dbe40a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128451919) * 33e4d2ef7f0db97a5fad81b5ef4542a299f805e6 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128453915) * 2f413575dd750406c2dfe2f04b93283ca629fbc9 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128456492) * 367beca92524d27a0fdbc918e8c080bb7957be3a : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129033648) * 2a3224847cee17db6416a4b86dfa3d1e394aca32 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129209349) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11
flinkbot edited a comment on issue #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11 URL: https://github.com/apache/flink/pull/9775#issuecomment-535327571 ## CI report: * 7ca40f4410862da362c16e5a9410ca692854451f : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129209325) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-534087366 ## CI report: * dd50b7a2af07a87280b2e358d31d0ad9345ad8c4 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743957) * abaf8ebcbd17e79ae92f9e66ed96772262919370 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129035121) * ee70e3344eda77f3d180b63460444f8e3064b2cb : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129151429) * 07a3ed3525cd48ee1e8c9d2349c48d0031ca167e : CANCELED [Build](https://travis-ci.com/flink-ci/flink/builds/129206839) * 541187ec0f55a5dd3b0e390320a45bb30755caa9 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129209308) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
flinkbot edited a comment on issue #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0 URL: https://github.com/apache/flink/pull/9712#issuecomment-532984287 ## CI report: * f2b7710f65f478342de389c8e099799287ddf3f9 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128285030) * 19872fd562f606e9738f5b5a77f7a1d3e93a39b9 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128302786) * 211f984c5380d61cf78f94e8ddcb4576f5c3f751 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128442800) * ed7ead402b73f1d5958f13c050dcda4b22dbe40a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128451919) * 33e4d2ef7f0db97a5fad81b5ef4542a299f805e6 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/128453915) * 2f413575dd750406c2dfe2f04b93283ca629fbc9 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128456492) * 367beca92524d27a0fdbc918e8c080bb7957be3a : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129033648) * 2a3224847cee17db6416a4b86dfa3d1e394aca32 : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9688: [FLINK-13056][runtime] Introduce FastRestartPipelinedRegionStrategy
flinkbot edited a comment on issue #9688: [FLINK-13056][runtime] Introduce FastRestartPipelinedRegionStrategy URL: https://github.com/apache/flink/pull/9688#issuecomment-531737820 ## CI report: * 23278b65f27154ae066841334f0d0b06885e713a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/127786107) * 5831d972c3f2a68398438015352e5a5e0a8de9da : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129102026) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot commented on issue #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11
flinkbot commented on issue #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11 URL: https://github.com/apache/flink/pull/9775#issuecomment-535327571 ## CI report: * 7ca40f4410862da362c16e5a9410ca692854451f : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi…
flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi… URL: https://github.com/apache/flink/pull/9773#issuecomment-535013943 ## CI report: * a7a643c54c912e9a42b1dbf887f14eba2ac80f82 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129101985) * c0f46541c04433aaef6301b81da9b34cc1d304a8 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129208037) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-534879961 ## CI report: * 8fffbbc4bc605ccedad7412ac2292d3488261fa3 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129046428) * 120f12f13987874b1738da7db1aa370f4335830a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129105662) * 8d3ff9abe90e8e23b2bd1ec12cd99dc7444f6a8d : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129205704) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-534087366 ## CI report: * dd50b7a2af07a87280b2e358d31d0ad9345ad8c4 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743957) * abaf8ebcbd17e79ae92f9e66ed96772262919370 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129035121) * ee70e3344eda77f3d180b63460444f8e3064b2cb : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129151429) * 07a3ed3525cd48ee1e8c9d2349c48d0031ca167e : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129206839) * 541187ec0f55a5dd3b0e390320a45bb30755caa9 : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-14019) Python environment and dependency management
[ https://issues.apache.org/jira/browse/FLINK-14019?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938254#comment-16938254 ] Wei Zhong commented on FLINK-14019: --- Thanks for your reply [~sunjincheng121]! I have send the discussion mail to d...@flink.apache.org, please take a look. :) > Python environment and dependency management > > > Key: FLINK-14019 > URL: https://issues.apache.org/jira/browse/FLINK-14019 > Project: Flink > Issue Type: Sub-task > Components: API / Python >Reporter: Dian Fu >Assignee: Wei Zhong >Priority: Major > Fix For: 1.10.0 > > > A Python user-defined functions may depend on third party dependencies. We > should provide a proper way to handle it: > # Provide a way to let users specifying the dependencies > # Provide a way to let users specifying the Python used -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] dianfu commented on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
dianfu commented on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-535325224 @hequn8128 Thanks for the latest comments. Updated. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-14059) Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
[
https://issues.apache.org/jira/browse/FLINK-14059?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zhu Zhu updated FLINK-14059:
Component/s: Table SQL / Runtime
Runtime / Coordination
> Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
> ---
>
> Key: FLINK-14059
> URL: https://issues.apache.org/jira/browse/FLINK-14059
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Coordination, Table SQL / Runtime
>Affects Versions: 1.10.0
>Reporter: Xintong Song
>Assignee: Zhu Zhu
>Priority: Major
> Fix For: 1.10.0
>
>
> * Introduce option {{allSourcesInSamePipelinedRegion}} in {{ExecutionConfig}}
> * Set it to {{true}} by default
> * Set it to {{false}} for SQL/Table API bounded batch jobs by the Blink
> planner
> This step should not introduce any behavior changes.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-14059) Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
[
https://issues.apache.org/jira/browse/FLINK-14059?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zhu Zhu updated FLINK-14059:
Fix Version/s: 1.10.0
> Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
> ---
>
> Key: FLINK-14059
> URL: https://issues.apache.org/jira/browse/FLINK-14059
> Project: Flink
> Issue Type: Sub-task
>Affects Versions: 1.10.0
>Reporter: Xintong Song
>Assignee: Zhu Zhu
>Priority: Major
> Fix For: 1.10.0
>
>
> * Introduce option {{allSourcesInSamePipelinedRegion}} in {{ExecutionConfig}}
> * Set it to {{true}} by default
> * Set it to {{false}} for SQL/Table API bounded batch jobs by the Blink
> planner
> This step should not introduce any behavior changes.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-14059) Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
[
https://issues.apache.org/jira/browse/FLINK-14059?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Zhu Zhu updated FLINK-14059:
Affects Version/s: 1.10.0
> Introduce option allSourcesInSamePipelinedRegion in ExecutionConfig
> ---
>
> Key: FLINK-14059
> URL: https://issues.apache.org/jira/browse/FLINK-14059
> Project: Flink
> Issue Type: Sub-task
>Affects Versions: 1.10.0
>Reporter: Xintong Song
>Assignee: Zhu Zhu
>Priority: Major
>
> * Introduce option {{allSourcesInSamePipelinedRegion}} in {{ExecutionConfig}}
> * Set it to {{true}} by default
> * Set it to {{false}} for SQL/Table API bounded batch jobs by the Blink
> planner
> This step should not introduce any behavior changes.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938252#comment-16938252
]
Jingsong Lee commented on FLINK-13740:
--
For #1, I mean BinaryGeneric still extends BinaryFormat and has javaObject, it
still lazy. The only difference is that we need do these materialize with pass
javaObjectSer to it.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328424222
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonScalarFunctionSplitRule.scala
##
@@ -0,0 +1,219 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode, RexProgram}
+import org.apache.calcite.sql.validate.SqlValidatorUtil
+import org.apache.flink.table.functions.FunctionLanguage
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.utils.ScalarSqlFunction
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc
+import org.apache.flink.table.plan.util.PythonUtil.containsFunctionOf
+import org.apache.flink.table.plan.util.{InputRefVisitor, RexDefaultVisitor}
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s.
After this rule
+ * is applied, each [[FlinkLogicalCalc]] will only contain Python
[[ScalarFunction]]s or Java
+ * [[ScalarFunction]]s. This is to ensure that the Python [[ScalarFunction]]s
which could be
+ * executed in a batch are grouped into the same [[FlinkLogicalCalc]] node.
+ */
+class PythonScalarFunctionSplitRule extends RelOptRule(
+ operand(classOf[FlinkLogicalCalc], any),
+ "PythonScalarFunctionSplitRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val program = calc.getProgram
+
+// This rule matches if one of the following cases is met:
+// 1. There are Python functions and Java functions mixed in the Calc
+// 2. There are Python functions in the condition of the Calc
+(program.getExprList.exists(containsFunctionOf(_,
FunctionLanguage.PYTHON)) &&
+ program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.JVM)))
||
+Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .exists(containsFunctionOf(_, FunctionLanguage.PYTHON))
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val input = calc.getInput
+val rexBuilder = call.builder().getRexBuilder
+val program = calc.getProgram
+val extractedRexCalls = new mutable.ArrayBuffer[RexCall]()
+
+val convertPythonFunction =
+ program.getProjectList
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive =
false)) ||
+ Option(program.getCondition)
+.map(program.expandLocalRef)
+.exists(expr =>
+ containsFunctionOf(expr, FunctionLanguage.JVM, recursive = false) ||
+containsFunctionOf(expr, FunctionLanguage.PYTHON))
+
+val extractedFunctionOffset = input.getRowType.getFieldCount
+val splitter = new ScalarFunctionSplitter(
+ extractedFunctionOffset,
+ extractedRexCalls,
+ convertPythonFunction)
+
+val newProjects = program.getProjectList
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val newCondition = Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val accessedFields = extractRefInputFields(
Review comment:
One line for this?
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328422873
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala
##
@@ -233,8 +241,9 @@ object FlinkRuleSets {
StreamTableSourceScanRule.INSTANCE,
DataStreamMatchRule.INSTANCE,
DataStreamTableAggregateRule.INSTANCE,
-DataStreamGroupWindowTableAggregateRule.INSTANCE
- )
+DataStreamGroupWindowTableAggregateRule.INSTANCE,
+DataStreamPythonCalcRule.INSTANCE
+)
Review comment:
Indent the `)`
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328426233
##
File path:
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/plan/PythonScalarFunctionSplitRuleTest.scala
##
@@ -0,0 +1,238 @@
+/*
+ * 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.table.plan
+
+import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
+import org.apache.flink.api.scala._
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.functions.{FunctionLanguage, ScalarFunction}
+import org.apache.flink.table.utils.TableTestUtil._
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Test
+
+class PythonScalarFunctionSplitRuleTest extends TableTestBase {
+
+ @Test
+ def testPythonFunctionAsInputOfJavaFunction(): Unit = {
+val util = streamTestUtil()
+val table = util.addTable[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
+util.tableEnv.registerFunction("pyFunc1", new
PythonScalarFunction("pyFunc1"))
+
+val resultTable = table
+ .select("pyFunc1(a, b) + 1")
+
+val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+"DataStreamPythonCalc",
+streamTableNode(table),
+term("select", "pyFunc1(a, b) AS f0")
+ ),
+ term("select", "+(f0, 1) AS _c0")
+)
+
+util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testPythonFunctionMixedWithJavaFunction(): Unit = {
+val util = streamTestUtil()
+val table = util.addTable[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
+util.tableEnv.registerFunction("pyFunc1", new
PythonScalarFunction("pyFunc1"))
+
+val resultTable = table
+ .select("pyFunc1(a, b), c + 1")
+
+val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+"DataStreamPythonCalc",
+streamTableNode(table),
+term("select", "c", "pyFunc1(a, b) AS f0")
+ ),
+ term("select", "f0 AS _c0", "+(c, 1) AS _c1")
+)
+
+util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testPythonFunctionMixedWithJavaFunctionInWhereClause(): Unit = {
+val util = streamTestUtil()
+val table = util.addTable[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
+util.tableEnv.registerFunction("pyFunc1", new
PythonScalarFunction("pyFunc1"))
+util.tableEnv.registerFunction("pyFunc2", new
PythonScalarFunction("pyFunc2"))
+
+val resultTable = table
+ .where("pyFunc2(a, c) > 0")
+ .select("pyFunc1(a, b), c + 1")
+
+val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+"DataStreamPythonCalc",
+streamTableNode(table),
+term("select", "c", "pyFunc1(a, b) AS f0", "pyFunc2(a, c) AS f1")
+),
+ term("select", "f0 AS _c0", "+(c, 1) AS _c1"),
+ term("where", ">(f1, 0)")
+ )
+
+util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testPythonFunctionInWhereClause(): Unit = {
+val util = streamTestUtil()
+val table = util.addTable[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
+util.tableEnv.registerFunction("pyFunc1", new
PythonScalarFunction("pyFunc1"))
+util.tableEnv.registerFunction("pyFunc2", new
BooleanPythonScalarFunction("pyFunc2"))
+
+val resultTable = table
+ .where("pyFunc2(a, c)")
+ .select("pyFunc1(a, b)")
+
+val expected = unaryNode(
+ "DataStreamCalc",
+ unaryNode(
+"DataStreamPythonCalc",
+streamTableNode(table),
+term("select", "pyFunc1(a, b) AS f0", "pyFunc2(a, c) AS f1")
+ ),
+ term("select", "f0 AS _c0"),
+ term("where", "f1")
+)
+
+util.verifyTable(resultTable, expected)
+ }
+
+ @Test
+ def testChainingPythonFunction(): Unit = {
+val util = streamTestUtil()
+val table = util.addTable[(Int, Int, Int)]("MyTable", 'a, 'b, 'c)
+util.tableEnv.registerFunction("pyFunc1", new
PythonScalarFunction("pyFunc1"))
+util.tableEnv.registerFunction("pyFunc2", new
PythonScalarFunction("pyFunc2"))
+util.tableEnv.registerFunction("pyFunc3", new
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328425219
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonScalarFunctionSplitRule.scala
##
@@ -0,0 +1,219 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode, RexProgram}
+import org.apache.calcite.sql.validate.SqlValidatorUtil
+import org.apache.flink.table.functions.FunctionLanguage
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.utils.ScalarSqlFunction
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc
+import org.apache.flink.table.plan.util.PythonUtil.containsFunctionOf
+import org.apache.flink.table.plan.util.{InputRefVisitor, RexDefaultVisitor}
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s.
After this rule
+ * is applied, each [[FlinkLogicalCalc]] will only contain Python
[[ScalarFunction]]s or Java
+ * [[ScalarFunction]]s. This is to ensure that the Python [[ScalarFunction]]s
which could be
+ * executed in a batch are grouped into the same [[FlinkLogicalCalc]] node.
+ */
+class PythonScalarFunctionSplitRule extends RelOptRule(
+ operand(classOf[FlinkLogicalCalc], any),
+ "PythonScalarFunctionSplitRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val program = calc.getProgram
+
+// This rule matches if one of the following cases is met:
+// 1. There are Python functions and Java functions mixed in the Calc
+// 2. There are Python functions in the condition of the Calc
+(program.getExprList.exists(containsFunctionOf(_,
FunctionLanguage.PYTHON)) &&
+ program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.JVM)))
||
+Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .exists(containsFunctionOf(_, FunctionLanguage.PYTHON))
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val input = calc.getInput
+val rexBuilder = call.builder().getRexBuilder
+val program = calc.getProgram
+val extractedRexCalls = new mutable.ArrayBuffer[RexCall]()
+
+val convertPythonFunction =
+ program.getProjectList
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive =
false)) ||
+ Option(program.getCondition)
+.map(program.expandLocalRef)
+.exists(expr =>
+ containsFunctionOf(expr, FunctionLanguage.JVM, recursive = false) ||
+containsFunctionOf(expr, FunctionLanguage.PYTHON))
+
+val extractedFunctionOffset = input.getRowType.getFieldCount
+val splitter = new ScalarFunctionSplitter(
+ extractedFunctionOffset,
+ extractedRexCalls,
+ convertPythonFunction)
+
+val newProjects = program.getProjectList
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val newCondition = Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val accessedFields = extractRefInputFields(
+ newProjects, newCondition, extractedFunctionOffset)
+
+val bottomCalcProjects =
+ accessedFields.map(RexInputRef.of(_, input.getRowType)) ++
extractedRexCalls
+val bottomCalcFieldNames = SqlValidatorUtil.uniquify(
+ accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++
+extractedRexCalls.indices.map("f" + _),
+ rexBuilder.getTypeFactory.getTypeSystem.isSchemaCaseSensitive)
+
+val bottomCalc = new FlinkLogicalCalc(
+
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328425098
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonScalarFunctionSplitRule.scala
##
@@ -0,0 +1,219 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode, RexProgram}
+import org.apache.calcite.sql.validate.SqlValidatorUtil
+import org.apache.flink.table.functions.FunctionLanguage
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.utils.ScalarSqlFunction
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc
+import org.apache.flink.table.plan.util.PythonUtil.containsFunctionOf
+import org.apache.flink.table.plan.util.{InputRefVisitor, RexDefaultVisitor}
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s.
After this rule
+ * is applied, each [[FlinkLogicalCalc]] will only contain Python
[[ScalarFunction]]s or Java
+ * [[ScalarFunction]]s. This is to ensure that the Python [[ScalarFunction]]s
which could be
+ * executed in a batch are grouped into the same [[FlinkLogicalCalc]] node.
+ */
+class PythonScalarFunctionSplitRule extends RelOptRule(
+ operand(classOf[FlinkLogicalCalc], any),
+ "PythonScalarFunctionSplitRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val program = calc.getProgram
+
+// This rule matches if one of the following cases is met:
+// 1. There are Python functions and Java functions mixed in the Calc
+// 2. There are Python functions in the condition of the Calc
+(program.getExprList.exists(containsFunctionOf(_,
FunctionLanguage.PYTHON)) &&
+ program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.JVM)))
||
+Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .exists(containsFunctionOf(_, FunctionLanguage.PYTHON))
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val input = calc.getInput
+val rexBuilder = call.builder().getRexBuilder
+val program = calc.getProgram
+val extractedRexCalls = new mutable.ArrayBuffer[RexCall]()
+
+val convertPythonFunction =
+ program.getProjectList
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive =
false)) ||
+ Option(program.getCondition)
+.map(program.expandLocalRef)
+.exists(expr =>
+ containsFunctionOf(expr, FunctionLanguage.JVM, recursive = false) ||
+containsFunctionOf(expr, FunctionLanguage.PYTHON))
+
+val extractedFunctionOffset = input.getRowType.getFieldCount
+val splitter = new ScalarFunctionSplitter(
+ extractedFunctionOffset,
+ extractedRexCalls,
+ convertPythonFunction)
+
+val newProjects = program.getProjectList
Review comment:
This part of code can be simplified as:
```
val newProjects =
program.getProjectList.map(program.expandLocalRef(_).accept(splitter))
val newCondition =
Option(program.getCondition).map(program.expandLocalRef(_).accept(splitter))
val accessedFields = extractRefInputFields(newProjects, newCondition,
extractedFunctionOffset)
```
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328425659
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/python/SimplePythonFunction.java
##
@@ -0,0 +1,56 @@
+/*
+ * 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.table.functions.python;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.Preconditions;
+
+/**
+ * A simple implementation of {@link PythonFunction}.
+ */
+@Internal
+public final class SimplePythonFunction implements PythonFunction {
Review comment:
Remove this class for now? It has not been used by any classes.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#discussion_r328422557 ## File path: flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/FlinkRuleSets.scala ## @@ -19,6 +19,7 @@ package org.apache.flink.table.plan.rules import org.apache.calcite.rel.core.RelFactories +import org.apache.calcite.rel.rules Review comment: Unnecessary change. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #9747: [FLINK-13985] Use unsafe memory for managed memory
xintongsong commented on a change in pull request #9747: [FLINK-13985] Use
unsafe memory for managed memory
URL: https://github.com/apache/flink/pull/9747#discussion_r328422193
##
File path:
flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
##
@@ -86,16 +86,16 @@ public static MemorySegment
allocateUnpooledOffHeapMemory(int size) {
}
/**
-* Allocates some unpooled off-heap memory and creates a new memory
segment that
-* represents that memory.
+* Allocates an unpooled off-heap memory and creates a new memory
segment to represent that memory.
*
* @param size The size of the off-heap memory segment to allocate.
* @param owner The owner to associate with the off-heap memory segment.
* @return A new memory segment, backed by unpooled off-heap memory.
*/
public static MemorySegment allocateUnpooledOffHeapMemory(int size,
Object owner) {
- ByteBuffer memory = ByteBuffer.allocateDirect(size);
- return new HybridMemorySegment(memory, owner);
+ long address = MemoryUtils.allocateUnsafe(size);
+ ByteBuffer offHeapBuffer =
MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);
+ return new HybridMemorySegment(offHeapBuffer, owner,
MemoryUtils.createMemoryGcCleaner(offHeapBuffer, address));
Review comment:
We intend to use unsafe memory for memory manager (off-heap managed memory)
only. This interface is also used by other components (`FileChannelBoundedData`
and `NetworkBufferPool`), where direct memory limited by
`-XX:MaxDirectMemorySize` is expected.
I think we should not change this interface, but add another
`allocateUnpooledUnsafeOffHeapMemory` and use it only in memory manager.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #9747: [FLINK-13985] Use unsafe memory for managed memory
xintongsong commented on a change in pull request #9747: [FLINK-13985] Use
unsafe memory for managed memory
URL: https://github.com/apache/flink/pull/9747#discussion_r328425366
##
File path:
flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java
##
@@ -57,4 +64,79 @@
/** Should not be instantiated. */
private MemoryUtils() {}
+
+ private static Constructor
getDirectBufferPrivateConstructor() {
+ //noinspection OverlyBroadCatchBlock
+ try {
+ Constructor constructor =
+
ByteBuffer.allocateDirect(1).getClass().getDeclaredConstructor(long.class,
int.class);
+ constructor.setAccessible(true);
+ return constructor;
+ } catch (NoSuchMethodException e) {
+ ExceptionUtils.rethrow(
+ e,
+ "The private constructor
java.nio.DirectByteBuffer.(long, int) is not available.");
+ } catch (SecurityException e) {
+ ExceptionUtils.rethrow(
+ e,
+ "The private constructor
java.nio.DirectByteBuffer.(long, int) is not available, " +
+ "permission denied by security
manager");
+ } catch (Throwable t) {
+ ExceptionUtils.rethrow(
+ t,
+ "Unclassified error while trying to access
private constructor " +
+ "java.nio.DirectByteBuffer.(long,
int).");
+ }
+ throw new RuntimeException("unexpected to avoid returning
null");
+ }
+
+ /**
+* Allocates unsafe native memory.
+*
+* @param size size of the unsafe memory to allocate.
+* @return address of the allocated unsafe memory
+*/
+ static long allocateUnsafe(long size) {
+ return UNSAFE.allocateMemory(Math.max(1L, size));
+ }
+
+ /**
+* Creates a cleaner to release the unsafe memory by VM GC.
+*
+* When memory owner becomes phantom reachable,
+* GC will release the underlying unsafe memory if not released yet.
+*
+* @param owner memory owner which phantom reaching is to monitor by GC
and release the unsafe memory
+* @param address address of the unsafe memory to release
+* @return action to run to release the unsafe memory manually
+*/
+ static Runnable createMemoryGcCleaner(@Nullable Object owner, long
address) {
+ return createGcCleaner(owner, () -> releaseUnsafe(address));
+ }
+
+ @SuppressWarnings("UseOfSunClasses")
+ private static Runnable createGcCleaner(@Nullable Object owner,
Runnable toClean) {
+ return owner == null ? toClean : sun.misc.Cleaner.create(owner,
toClean)::clean;
Review comment:
If I understand correctly, the reason we use `Cleaner#clean` instead of
`toClean` directly, is to make sure the cleaner is removed when it's invoked in
`HybridMemorySegment#free`, so there won't be duplicated `Unsafe#freeMemory`
calls in later GCs?
I think we should add a test case to make sure nobody breaks it in the
future.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #9747: [FLINK-13985] Use unsafe memory for managed memory
xintongsong commented on a change in pull request #9747: [FLINK-13985] Use
unsafe memory for managed memory
URL: https://github.com/apache/flink/pull/9747#discussion_r328413824
##
File path:
flink-core/src/test/java/org/apache/flink/core/memory/CrossSegmentTypeTest.java
##
@@ -44,15 +43,15 @@
@Test
public void testCompareBytesMixedSegments() {
MemorySegment[] segs1 = {
- new HeapMemorySegment(new byte[pageSize]),
- new HybridMemorySegment(new byte[pageSize]),
- new
HybridMemorySegment(ByteBuffer.allocateDirect(pageSize))
+ MemorySegmentFactory.allocateUnpooledSegment(pageSize),
Review comment:
We should not replace `new HeapMemorySegment()` with
`MemorySegementFactory.allocateUnpooledSegment`. The latter creates a
`HybridMemorySegment` instead of `HeapMemorySegment`, while this test case is
meant to also validate the behavior of `HeapMemorySegment`.
Same for other occurrences in this class and `OperationsOnFreedSegmentTest `.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] xintongsong commented on a change in pull request #9747: [FLINK-13985] Use unsafe memory for managed memory
xintongsong commented on a change in pull request #9747: [FLINK-13985] Use
unsafe memory for managed memory
URL: https://github.com/apache/flink/pull/9747#discussion_r328422844
##
File path:
flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
##
@@ -86,16 +86,16 @@ public static MemorySegment
allocateUnpooledOffHeapMemory(int size) {
}
/**
-* Allocates some unpooled off-heap memory and creates a new memory
segment that
-* represents that memory.
+* Allocates an unpooled off-heap memory and creates a new memory
segment to represent that memory.
*
* @param size The size of the off-heap memory segment to allocate.
* @param owner The owner to associate with the off-heap memory segment.
* @return A new memory segment, backed by unpooled off-heap memory.
*/
public static MemorySegment allocateUnpooledOffHeapMemory(int size,
Object owner) {
- ByteBuffer memory = ByteBuffer.allocateDirect(size);
- return new HybridMemorySegment(memory, owner);
+ long address = MemoryUtils.allocateUnsafe(size);
+ ByteBuffer offHeapBuffer =
MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);
+ return new HybridMemorySegment(offHeapBuffer, owner,
MemoryUtils.createMemoryGcCleaner(offHeapBuffer, address));
Review comment:
Both safe and unsafe off-heap memory can be wrapped in
`HybridMemorySegment`. The only difference is that the unsafe ones have
non-null cleaners.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhuzhurk commented on issue #9688: [FLINK-13056][runtime] Introduce FastRestartPipelinedRegionStrategy
zhuzhurk commented on issue #9688: [FLINK-13056][runtime] Introduce FastRestartPipelinedRegionStrategy URL: https://github.com/apache/flink/pull/9688#issuecomment-535323407 @flinkbot run travis 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh
flinkbot edited a comment on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh URL: https://github.com/apache/flink/pull/9774#issuecomment-535311583 ## CI report: * b863a8fcbbf97650009e0947b10f283e6954ace3 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129204536) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi…
flinkbot edited a comment on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi… URL: https://github.com/apache/flink/pull/9773#issuecomment-535013943 ## CI report: * a7a643c54c912e9a42b1dbf887f14eba2ac80f82 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129101985) * c0f46541c04433aaef6301b81da9b34cc1d304a8 : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938242#comment-16938242
]
Jark Wu commented on FLINK-13740:
-
I don't think #1 gets better performance, because it forces that we have to
materialize/serialize at the beginning of creating BinaryGeneric.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot commented on issue #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11
flinkbot commented on issue #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11 URL: https://github.com/apache/flink/pull/9775#issuecomment-535322691 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit 7ca40f4410862da362c16e5a9410ca692854451f (Thu Sep 26 03:59:24 UTC 2019) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-534087366 ## CI report: * dd50b7a2af07a87280b2e358d31d0ad9345ad8c4 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743957) * abaf8ebcbd17e79ae92f9e66ed96772262919370 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129035121) * ee70e3344eda77f3d180b63460444f8e3064b2cb : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129151429) * 07a3ed3525cd48ee1e8c9d2349c48d0031ca167e : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129206839) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-14186) Nightly e2e-misc-jdk11 fails complaining unrecognized VM option
[
https://issues.apache.org/jira/browse/FLINK-14186?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-14186:
---
Labels: pull-request-available test-stability (was: test-stability)
> Nightly e2e-misc-jdk11 fails complaining unrecognized VM option
> ---
>
> Key: FLINK-14186
> URL: https://issues.apache.org/jira/browse/FLINK-14186
> Project: Flink
> Issue Type: Bug
> Components: Tests
>Reporter: Yu Li
>Assignee: Yun Gao
>Priority: Critical
> Labels: pull-request-available, test-stability
>
> As titled, this failure reproduces stably in recent nightly runs, such as
> https://api.travis-ci.org/v3/job/588201921/log.txt . From the log we could
> see message like below:
> {noformat}
> Unrecognized VM option 'UseParNewGC'
> Error: Could not create the Java Virtual Machine.
> Error: A fatal exception has occurred. Program will exit.
> {noformat}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] gaoyunhaii opened a new pull request #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11
gaoyunhaii opened a new pull request #9775: [FLINK-14186][e2e] Let JVM ignores unrecoginzed options to avoid end to end test failure in es 2.3.5 with JDK11 URL: https://github.com/apache/flink/pull/9775 ## What is the purpose of the change This PR fixes the end-to-end test failure when testing elastic search link with ES 2.3.5 on JDK 11. Since Java has removed some GC options with the change on the GC algorithms, we need to let the JVM ignore the unrecognized options, otherwise ES 2.3.5 will fail to startup with JDK 11 due to it configures some default options that have been removed in JDK 11. ## Brief change log - 7ca40f4410862da362c16e5a9410ca692854451f configures JVM to ignore the unrecognized options so that when starting ES cluster. ## Verifying this change This change is tested by manually run run-nightly-tests.sh under flink-end-to-end-tests. ## 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: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: **no** - The S3 file system connector: **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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938238#comment-16938238
]
Jingsong Lee commented on FLINK-13740:
--
Let me clear the solutions:
1.Let BinaryGeneric not extends LazyBinaryFormat and all supported method of
BinaryGeneric should pass javaObjectSer to it.
2.Add sync to materialize(there are many invoking to it), and remove copy in
BinaryGeneric.
For performance, #1 is better.
For easy fix and LazyBinaryFormat definition, #2 is better.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] Mrart commented on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi…
Mrart commented on issue #9773: [FLINK-14210][metrics]support connect timeout and write timeout confi… URL: https://github.com/apache/flink/pull/9773#issuecomment-535320428 @1u0 ths for review 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Assigned] (FLINK-14080) Support precision of TimestampType
[ https://issues.apache.org/jira/browse/FLINK-14080?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jark Wu reassigned FLINK-14080: --- Assignee: Zhenghua Gao > Support precision of TimestampType > -- > > Key: FLINK-14080 > URL: https://issues.apache.org/jira/browse/FLINK-14080 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: Jingsong Lee >Assignee: Zhenghua Gao >Priority: Major > Fix For: 1.10.0 > > > Timestamp values have a precision in fractional seconds that range from 0 to > 9. For example, a precision of 0 means that no fractional seconds are stored, > 3 means that the timestamp stores milliseconds, and 9 means a precision of > nanoseconds. 0 is the minimum precision, and 9 is the maximum. > Now, blink planner represent Timestamp values as Long(milliseconds) and only > support precision is 3, we need support more precision Timestamp values. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-14080) Support precision of TimestampType
[ https://issues.apache.org/jira/browse/FLINK-14080?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938237#comment-16938237 ] Jingsong Lee commented on FLINK-14080: -- [~jark] Can you assign this ticket to [~docete] ? > Support precision of TimestampType > -- > > Key: FLINK-14080 > URL: https://issues.apache.org/jira/browse/FLINK-14080 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: Jingsong Lee >Assignee: Zhenghua Gao >Priority: Major > Fix For: 1.10.0 > > > Timestamp values have a precision in fractional seconds that range from 0 to > 9. For example, a precision of 0 means that no fractional seconds are stored, > 3 means that the timestamp stores milliseconds, and 9 means a precision of > nanoseconds. 0 is the minimum precision, and 9 is the maximum. > Now, blink planner represent Timestamp values as Long(milliseconds) and only > support precision is 3, we need support more precision Timestamp values. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-14186) Nightly e2e-misc-jdk11 fails complaining unrecognized VM option
[
https://issues.apache.org/jira/browse/FLINK-14186?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938235#comment-16938235
]
Yun Gao commented on FLINK-14186:
-
The reason of this issue is that -XX:+UseParNewGC [has been removed from the
JDK11|[https://bugs.openjdk.java.net/browse/JDK-8151084]] as the garbage
collector algorithms have evolved. However, Elastic Search 2.3.5 has
configured this option in its startup script. Therefore, the Elastic Search
2.3.5 will fail to start on JDK11, which cause the test fail.
To fix this problem, we can set environment variable
JAVA_OPTS="-XX:+IgnoreUnrecognizedVMOptions" when starting the Elastic Search
cluster. This environment variable parameter is provided by Elastic Search
2.3.5 script and IgnoreUnrecognizedVMOptions let JVM ignores unrecognized
options instead throw an error.
The Elastic Search 5.x and 6.x have removed -XX:+UseParNewGC from the default
JVM config and thus they should not met this issue. Besides, in these two
version JAVA_OPTS is deprecated and ES_JAVA_OPTS is used instead. Since the two
version currently do not have problem, I think we do not need to also configure
ES_JAVA_OPTS now.
For reference, the code involved in Flink is:
# flink-end-to-end-tests/run-nightly-tests.sh have three ElasticSearch related
tests. They all finally called
flink-end-to-end-tests/test-scripts/test_streaming_elasticsearch.sh.
# test_streaming_elasticsearch.sh further called the function
setup_elasticsearch included from
flink-end-to-end-tests/test-scripts/elasticsearch-common.sh
# setup_elasticsearch downloads the specified ElasticSearch distribution
package and unzip it. Then it calls "_$elasticsearchDir/bin/elasticsearch &"_
to start the __ ElasticSearch __ cluster. We should then modify the startup
environment here.
> Nightly e2e-misc-jdk11 fails complaining unrecognized VM option
> ---
>
> Key: FLINK-14186
> URL: https://issues.apache.org/jira/browse/FLINK-14186
> Project: Flink
> Issue Type: Bug
> Components: Tests
>Reporter: Yu Li
>Assignee: Yun Gao
>Priority: Critical
> Labels: test-stability
>
> As titled, this failure reproduces stably in recent nightly runs, such as
> https://api.travis-ci.org/v3/job/588201921/log.txt . From the log we could
> see message like below:
> {noformat}
> Unrecognized VM option 'UseParNewGC'
> Error: Could not create the Java Virtual Machine.
> Error: A fatal exception has occurred. Program will exit.
> {noformat}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-534879961 ## CI report: * 8fffbbc4bc605ccedad7412ac2292d3488261fa3 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129046428) * 120f12f13987874b1738da7db1aa370f4335830a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129105662) * 8d3ff9abe90e8e23b2bd1ec12cd99dc7444f6a8d : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129205704) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-12576) inputQueueLength metric does not work for LocalInputChannels
[
https://issues.apache.org/jira/browse/FLINK-12576?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938232#comment-16938232
]
Yingjie Cao commented on FLINK-12576:
-
[~alpinegizmo] I tried the instruction gave above, but the problem did not
reproduce.
The instruction I used is:
!image-2019-09-26-11-36-06-027.png!
Here is my result:
!image-2019-09-26-11-34-24-878.png!
Input queue length of both the local and remote channel are not always zero.
Did I do something wrong?
> inputQueueLength metric does not work for LocalInputChannels
>
>
> Key: FLINK-12576
> URL: https://issues.apache.org/jira/browse/FLINK-12576
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Metrics, Runtime / Network
>Affects Versions: 1.6.4, 1.7.2, 1.8.0, 1.9.0
>Reporter: Piotr Nowojski
>Assignee: Aitozi
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.9.0
>
> Attachments: Screen Shot 2019-09-24 at 3.11.15 PM.png, Screen Shot
> 2019-09-24 at 3.13.05 PM.png, Screen Shot 2019-09-24 at 3.22.36 PM.png,
> Screen Shot 2019-09-24 at 3.22.53 PM.png,
> flink-1.8-2-single-slot-TMs-input.png,
> flink-1.8-2-single-slot-TMs-output.png, flink-1.8-input-subtasks.png,
> flink-1.8-output-subtasks.png, image-2019-09-26-11-34-24-878.png,
> image-2019-09-26-11-36-06-027.png
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Currently {{inputQueueLength}} ignores LocalInputChannels
> ({{SingleInputGate#getNumberOfQueuedBuffers}}). This can can cause mistakes
> when looking for causes of back pressure (If task is back pressuring whole
> Flink job, but there is a data skew and only local input channels are being
> used).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
flinkbot edited a comment on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-534087366 ## CI report: * dd50b7a2af07a87280b2e358d31d0ad9345ad8c4 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/128743957) * abaf8ebcbd17e79ae92f9e66ed96772262919370 : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129035121) * ee70e3344eda77f3d180b63460444f8e3064b2cb : SUCCESS [Build](https://travis-ci.com/flink-ci/flink/builds/129151429) * 07a3ed3525cd48ee1e8c9d2349c48d0031ca167e : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938231#comment-16938231
]
Jark Wu commented on FLINK-13740:
-
#1, It will be a great regression for generic types. (most users are using
generic type for UDFs)
#2, I'm not sure whether it works. Because the javaObjectSer is mainly used for
materialize(), but materialize() is an interface of LazyBinaryFormat. It's
hard to pass serializer for a nested generic object.
An easy way maybe we can make operations of javaObjectSer in a synchronized
block, including:
1) {{InstantiationUtil.serializeToByteArray(javaObjectSer, javaObject)}} in
{{materialize()}}
2) {{javaObjectSer.copy(javaObject)}} in {{copy()}}
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-12576) inputQueueLength metric does not work for LocalInputChannels
[
https://issues.apache.org/jira/browse/FLINK-12576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yingjie Cao updated FLINK-12576:
Attachment: image-2019-09-26-11-36-06-027.png
> inputQueueLength metric does not work for LocalInputChannels
>
>
> Key: FLINK-12576
> URL: https://issues.apache.org/jira/browse/FLINK-12576
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Metrics, Runtime / Network
>Affects Versions: 1.6.4, 1.7.2, 1.8.0, 1.9.0
>Reporter: Piotr Nowojski
>Assignee: Aitozi
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.9.0
>
> Attachments: Screen Shot 2019-09-24 at 3.11.15 PM.png, Screen Shot
> 2019-09-24 at 3.13.05 PM.png, Screen Shot 2019-09-24 at 3.22.36 PM.png,
> Screen Shot 2019-09-24 at 3.22.53 PM.png,
> flink-1.8-2-single-slot-TMs-input.png,
> flink-1.8-2-single-slot-TMs-output.png, flink-1.8-input-subtasks.png,
> flink-1.8-output-subtasks.png, image-2019-09-26-11-34-24-878.png,
> image-2019-09-26-11-36-06-027.png
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Currently {{inputQueueLength}} ignores LocalInputChannels
> ({{SingleInputGate#getNumberOfQueuedBuffers}}). This can can cause mistakes
> when looking for causes of back pressure (If task is back pressuring whole
> Flink job, but there is a data skew and only local input channels are being
> used).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Updated] (FLINK-12576) inputQueueLength metric does not work for LocalInputChannels
[
https://issues.apache.org/jira/browse/FLINK-12576?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yingjie Cao updated FLINK-12576:
Attachment: image-2019-09-26-11-34-24-878.png
> inputQueueLength metric does not work for LocalInputChannels
>
>
> Key: FLINK-12576
> URL: https://issues.apache.org/jira/browse/FLINK-12576
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Metrics, Runtime / Network
>Affects Versions: 1.6.4, 1.7.2, 1.8.0, 1.9.0
>Reporter: Piotr Nowojski
>Assignee: Aitozi
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.9.0
>
> Attachments: Screen Shot 2019-09-24 at 3.11.15 PM.png, Screen Shot
> 2019-09-24 at 3.13.05 PM.png, Screen Shot 2019-09-24 at 3.22.36 PM.png,
> Screen Shot 2019-09-24 at 3.22.53 PM.png,
> flink-1.8-2-single-slot-TMs-input.png,
> flink-1.8-2-single-slot-TMs-output.png, flink-1.8-input-subtasks.png,
> flink-1.8-output-subtasks.png, image-2019-09-26-11-34-24-878.png
>
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Currently {{inputQueueLength}} ignores LocalInputChannels
> ({{SingleInputGate#getNumberOfQueuedBuffers}}). This can can cause mistakes
> when looking for causes of back pressure (If task is back pressuring whole
> Flink job, but there is a data skew and only local input channels are being
> used).
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime] Reducing synchronization in AsyncWaitOperator
zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime]
Reducing synchronization in AsyncWaitOperator
URL: https://github.com/apache/flink/pull/9717#discussion_r328422187
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/UnorderedStreamElementQueue.java
##
@@ -19,192 +19,146 @@
package org.apache.flink.streaming.api.operators.async.queue;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.streaming.api.operators.async.OperatorActions;
+import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.streaming.api.operators.TimestampedCollector;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamElement;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayDeque;
-import java.util.Arrays;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Deque;
import java.util.HashSet;
-import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.Queue;
import java.util.Set;
-import java.util.concurrent.Executor;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
/**
- * Unordered implementation of the {@link StreamElementQueue}. The unordered
stream element queue
- * emits asynchronous results as soon as they are completed. Additionally it
maintains the
- * watermark-stream record order. This means that no stream record can be
overtaken by a watermark
- * and no watermark can overtake a stream record. However, stream records
falling in the same
- * segment between two watermarks can overtake each other (their emission
order is not guaranteed).
+ * Unordered implementation of the {@link StreamElementQueue}. The unordered
stream element queue provides
+ * asynchronous results as soon as they are completed. Additionally it
maintains the watermark-stream record order.
Review comment:
Yes. Elements within a stage could overtake each other. One more thing for
further confirmation: how to guarantee that two stages can not overtake each
other?
e.g. assuming we have four stages : {a, b, c} {watermark1} {d, e, f}
{watermark2}, and the first stage is the {a, b, c}. If the user function
completes the element `e` in third stage firstly, that means the collection of
`e` would be emitted to output before the first stage? Or I missed something
else.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-12662) Record failure and restart information for ExecutionGraph to keep history information
[ https://issues.apache.org/jira/browse/FLINK-12662?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938225#comment-16938225 ] vinoyang commented on FLINK-12662: -- [~trohrmann] OK, I will wait until you are free. > Record failure and restart information for ExecutionGraph to keep history > information > - > > Key: FLINK-12662 > URL: https://issues.apache.org/jira/browse/FLINK-12662 > Project: Flink > Issue Type: Improvement > Components: Runtime / REST >Reporter: Ralph Su >Assignee: vinoyang >Priority: Major > > Currently > [https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/historyserver.html] > only show the completed jobs (completd, cancel, failed). Not showing any > intermediate failover. > Which make the cluster administrator/developer hard to find first place if > there is two failover happens. Feature ask is to > - make a failover as a record in history server as well. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-14019) Python environment and dependency management
[ https://issues.apache.org/jira/browse/FLINK-14019?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938224#comment-16938224 ] sunjincheng commented on FLINK-14019: - Thanks for the detail design doc [~zhongwei]! Could you please bring up the ML discussion then we can get more input about the Design. :) Thanks! > Python environment and dependency management > > > Key: FLINK-14019 > URL: https://issues.apache.org/jira/browse/FLINK-14019 > Project: Flink > Issue Type: Sub-task > Components: API / Python >Reporter: Dian Fu >Assignee: Wei Zhong >Priority: Major > Fix For: 1.10.0 > > > A Python user-defined functions may depend on third party dependencies. We > should provide a proper way to handle it: > # Provide a way to let users specifying the dependencies > # Provide a way to let users specifying the Python used -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh
flinkbot edited a comment on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh URL: https://github.com/apache/flink/pull/9774#issuecomment-535311583 ## CI report: * b863a8fcbbf97650009e0947b10f283e6954ace3 : PENDING [Build](https://travis-ci.com/flink-ci/flink/builds/129204536) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
flinkbot edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-534879961 ## CI report: * 8fffbbc4bc605ccedad7412ac2292d3488261fa3 : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129046428) * 120f12f13987874b1738da7db1aa370f4335830a : FAILURE [Build](https://travis-ci.com/flink-ci/flink/builds/129105662) * 8d3ff9abe90e8e23b2bd1ec12cd99dc7444f6a8d : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] dianfu commented on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
dianfu commented on issue #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution URL: https://github.com/apache/flink/pull/9748#issuecomment-535315320 @hequn8128 Thanks for your review. Updated. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] dianfu commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
dianfu commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328420024
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonScalarFunctionSplitRule.scala
##
@@ -0,0 +1,211 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode, RexProgram}
+import org.apache.calcite.sql.validate.SqlValidatorUtil
+import org.apache.flink.table.functions.FunctionLanguage
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.utils.ScalarSqlFunction
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc
+import org.apache.flink.table.plan.util.PythonUtil.containsFunctionOf
+import org.apache.flink.table.plan.util.{InputRefVisitor, RexDefaultVisitor}
+import
org.apache.flink.table.plan.rules.logical.PythonScalarFunctionSplitRule.extractRefInputFields
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s.
After this rule
+ * is applied, each [[FlinkLogicalCalc]] will only contain Python
[[ScalarFunction]]s or Java
+ * [[ScalarFunction]]s. This is to ensure that the Python [[ScalarFunction]]s
which could be
+ * executed in a batch are grouped into the same [[FlinkLogicalCalc]] node.
+ */
+class PythonScalarFunctionSplitRule extends RelOptRule(
+ operand(classOf[FlinkLogicalCalc], any),
+ "PythonScalarFunctionSplitRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val program = calc.getProgram
+program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.PYTHON))
&&
+program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.JVM))
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val input = calc.getInput
+val rexBuilder = call.builder().getRexBuilder
+val program = calc.getProgram
+val extractedRexCalls = new mutable.ArrayBuffer[RexCall]()
+
+val outerCallContainsJavaFunction =
+ program.getProjectList
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive =
false)) ||
+ Option(program.getCondition)
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive = false))
+
+val extractedFunctionOffset = input.getRowType.getFieldCount
+val splitter = new ScalarFunctionSplitter(
+ extractedFunctionOffset,
+ extractedRexCalls,
+ outerCallContainsJavaFunction)
+
+val newProjects = program.getProjectList
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val newCondition = Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val accessedFields = extractRefInputFields(
+ newProjects, newCondition, extractedFunctionOffset)
+
+val bottomCalcProjects =
+ accessedFields.map(RexInputRef.of(_, input.getRowType)) ++
extractedRexCalls
+val bottomCalcFieldNames = SqlValidatorUtil.uniquify(
+ accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++
+extractedRexCalls.indices.map("f" + _),
+ rexBuilder.getTypeFactory.getTypeSystem.isSchemaCaseSensitive)
+
+val bottomCalc = new FlinkLogicalCalc(
Review comment:
Good catch. Fixed.
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.
For queries about this service, please contact
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938206#comment-16938206
]
Jingsong Lee commented on FLINK-13740:
--
1.Let BinaryGeneric not extends LazyBinaryFormat
2.All supported method of BinaryGeneric should pass javaObjectSer to it.
It is a special LazyBinaryFormat, but it has different operation modes.
What do you think?
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938203#comment-16938203
]
Jark Wu commented on FLINK-13740:
-
But LazyBinaryFormat BinaryGeneric need the javaObjectSer...
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot commented on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh
flinkbot commented on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh URL: https://github.com/apache/flink/pull/9774#issuecomment-535311583 ## CI report: * b863a8fcbbf97650009e0947b10f283e6954ace3 : UNKNOWN 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938201#comment-16938201
]
Jingsong Lee commented on FLINK-13740:
--
Yes, A thorough solution is BinaryGeneric should never hold javaObjectSer.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot commented on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh
flinkbot commented on issue #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh URL: https://github.com/apache/flink/pull/9774#issuecomment-535310464 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit b863a8fcbbf97650009e0947b10f283e6954ace3 (Thu Sep 26 02:58:10 UTC 2019) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime] Reducing synchronization in AsyncWaitOperator
zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime]
Reducing synchronization in AsyncWaitOperator
URL: https://github.com/apache/flink/pull/9717#discussion_r328416725
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/execution/MailboxProcessor.java
##
@@ -200,7 +200,7 @@ private boolean processMail(TaskMailbox mailbox) throws
MailboxStateException, I
try {
maybeLetter.get().run();
} catch (Exception e) {
- e.printStackTrace();
+ throw new IllegalStateException("Cannot process
mail " + maybeLetter.get(), e);
Review comment:
Actually I concerned two issues:
- `IllegalStateException` seems not very accurate here.
- If we want to provide more message for debugging, maybe we could use
`ExceptionUtils.rethrow(e, "Cannot process mail " + maybeLetter.get());`. But I
am not sure whether `maybeLetter.get()` could give any helpful info here. If
the debug info is useless, it seems simple to remove `try...catch` directly.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] TisonKun opened a new pull request #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh
TisonKun opened a new pull request #9774: [hotfix][tools] Update comments in verify_scala_suffixes.sh URL: https://github.com/apache/flink/pull/9774 ## What is the purpose of the change flink-storm has been dropped. Correspondingly update comments. ## Verifying this change This change is a trivial rework / code cleanup without any test coverage. ## 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: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no) - The S3 file system connector: (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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime] Reducing synchronization in AsyncWaitOperator
zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime]
Reducing synchronization in AsyncWaitOperator
URL: https://github.com/apache/flink/pull/9717#discussion_r328416725
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/execution/MailboxProcessor.java
##
@@ -200,7 +200,7 @@ private boolean processMail(TaskMailbox mailbox) throws
MailboxStateException, I
try {
maybeLetter.get().run();
} catch (Exception e) {
- e.printStackTrace();
+ throw new IllegalStateException("Cannot process
mail " + maybeLetter.get(), e);
Review comment:
Actually I concerned two issues here:
- `IllegalStateException` seems not very accurate here.
- If we want to provide more message for debugging, maybe we could use
`ExceptionUtils.rethrow(e, "Cannot process mail " + maybeLetter.get());`. But I
am not sure whether `maybeLetter.get()` could give any helpful info here. If
the debug info is useless, it seems simple to remove `try...catch` directly.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328413950
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonScalarFunctionSplitRule.scala
##
@@ -0,0 +1,211 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode, RexProgram}
+import org.apache.calcite.sql.validate.SqlValidatorUtil
+import org.apache.flink.table.functions.FunctionLanguage
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.utils.ScalarSqlFunction
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc
+import org.apache.flink.table.plan.util.PythonUtil.containsFunctionOf
+import org.apache.flink.table.plan.util.{InputRefVisitor, RexDefaultVisitor}
+import
org.apache.flink.table.plan.rules.logical.PythonScalarFunctionSplitRule.extractRefInputFields
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s.
After this rule
+ * is applied, each [[FlinkLogicalCalc]] will only contain Python
[[ScalarFunction]]s or Java
+ * [[ScalarFunction]]s. This is to ensure that the Python [[ScalarFunction]]s
which could be
+ * executed in a batch are grouped into the same [[FlinkLogicalCalc]] node.
+ */
+class PythonScalarFunctionSplitRule extends RelOptRule(
+ operand(classOf[FlinkLogicalCalc], any),
+ "PythonScalarFunctionSplitRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val program = calc.getProgram
+program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.PYTHON))
&&
+program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.JVM))
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val input = calc.getInput
+val rexBuilder = call.builder().getRexBuilder
+val program = calc.getProgram
+val extractedRexCalls = new mutable.ArrayBuffer[RexCall]()
+
+val outerCallContainsJavaFunction =
+ program.getProjectList
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive =
false)) ||
+ Option(program.getCondition)
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive = false))
+
+val extractedFunctionOffset = input.getRowType.getFieldCount
+val splitter = new ScalarFunctionSplitter(
+ extractedFunctionOffset,
+ extractedRexCalls,
+ outerCallContainsJavaFunction)
+
+val newProjects = program.getProjectList
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val newCondition = Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val accessedFields = extractRefInputFields(
+ newProjects, newCondition, extractedFunctionOffset)
+
+val bottomCalcProjects =
+ accessedFields.map(RexInputRef.of(_, input.getRowType)) ++
extractedRexCalls
+val bottomCalcFieldNames = SqlValidatorUtil.uniquify(
+ accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++
+extractedRexCalls.indices.map("f" + _),
+ rexBuilder.getTypeFactory.getTypeSystem.isSchemaCaseSensitive)
+
+val bottomCalc = new FlinkLogicalCalc(
Review comment:
It seems we always treat the bottomCalc as the Calc contains python UDFs and
the top Calc contains java UDFs. However, in some cases, it would be the
opposite, i.e., when outerCallContainsJavaFunction is false.
The error could be reproduced by the following test:
```
@Test
def
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328408371
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/rules/logical/PythonScalarFunctionSplitRule.scala
##
@@ -0,0 +1,211 @@
+/*
+ * 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.table.plan.rules.logical
+
+import org.apache.calcite.plan.RelOptRule.{any, operand}
+import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
+import org.apache.calcite.rex.{RexCall, RexInputRef, RexNode, RexProgram}
+import org.apache.calcite.sql.validate.SqlValidatorUtil
+import org.apache.flink.table.functions.FunctionLanguage
+import org.apache.flink.table.functions.ScalarFunction
+import org.apache.flink.table.functions.utils.ScalarSqlFunction
+import org.apache.flink.table.plan.nodes.logical.FlinkLogicalCalc
+import org.apache.flink.table.plan.util.PythonUtil.containsFunctionOf
+import org.apache.flink.table.plan.util.{InputRefVisitor, RexDefaultVisitor}
+import
org.apache.flink.table.plan.rules.logical.PythonScalarFunctionSplitRule.extractRefInputFields
+
+import scala.collection.JavaConverters._
+import scala.collection.JavaConversions._
+import scala.collection.mutable
+
+/**
+ * Rule that splits [[FlinkLogicalCalc]] into multiple [[FlinkLogicalCalc]]s.
After this rule
+ * is applied, each [[FlinkLogicalCalc]] will only contain Python
[[ScalarFunction]]s or Java
+ * [[ScalarFunction]]s. This is to ensure that the Python [[ScalarFunction]]s
which could be
+ * executed in a batch are grouped into the same [[FlinkLogicalCalc]] node.
+ */
+class PythonScalarFunctionSplitRule extends RelOptRule(
+ operand(classOf[FlinkLogicalCalc], any),
+ "PythonScalarFunctionSplitRule") {
+
+ override def matches(call: RelOptRuleCall): Boolean = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val program = calc.getProgram
+program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.PYTHON))
&&
+program.getExprList.exists(containsFunctionOf(_, FunctionLanguage.JVM))
+ }
+
+ override def onMatch(call: RelOptRuleCall): Unit = {
+val calc: FlinkLogicalCalc = call.rel(0).asInstanceOf[FlinkLogicalCalc]
+val input = calc.getInput
+val rexBuilder = call.builder().getRexBuilder
+val program = calc.getProgram
+val extractedRexCalls = new mutable.ArrayBuffer[RexCall]()
+
+val outerCallContainsJavaFunction =
+ program.getProjectList
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive =
false)) ||
+ Option(program.getCondition)
+.map(program.expandLocalRef)
+.exists(containsFunctionOf(_, FunctionLanguage.JVM, recursive = false))
+
+val extractedFunctionOffset = input.getRowType.getFieldCount
+val splitter = new ScalarFunctionSplitter(
+ extractedFunctionOffset,
+ extractedRexCalls,
+ outerCallContainsJavaFunction)
+
+val newProjects = program.getProjectList
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val newCondition = Option(program.getCondition)
+ .map(program.expandLocalRef)
+ .map(_.accept(splitter))
+
+val accessedFields = extractRefInputFields(
+ newProjects, newCondition, extractedFunctionOffset)
+
+val bottomCalcProjects =
+ accessedFields.map(RexInputRef.of(_, input.getRowType)) ++
extractedRexCalls
+val bottomCalcFieldNames = SqlValidatorUtil.uniquify(
+ accessedFields.map(i => input.getRowType.getFieldNames.get(i)).toSeq ++
+extractedRexCalls.indices.map("f" + _),
+ rexBuilder.getTypeFactory.getTypeSystem.isSchemaCaseSensitive)
+
+val bottomCalc = new FlinkLogicalCalc(
+ calc.getCluster,
+ calc.getTraitSet,
+ input,
+ RexProgram.create(
+input.getRowType,
+bottomCalcProjects.toList,
+null,
+bottomCalcFieldNames,
+rexBuilder))
+
+val inputRewriter = new ExtractedFunctionInputRewriter(
+ extractedFunctionOffset,
[GitHub] [flink] hequn8128 commented on a change in pull request #9748: [FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for Python function execution
hequn8128 commented on a change in pull request #9748:
[FLINK-14016][python][flink-table-planner] Introduce DataStreamPythonCalc for
Python function execution
URL: https://github.com/apache/flink/pull/9748#discussion_r328404835
##
File path:
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamPythonCalc.scala
##
@@ -0,0 +1,67 @@
+/*
+ * 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.table.plan.nodes.datastream
+
+import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
+import org.apache.calcite.rel.RelNode
+import org.apache.calcite.rel.core.Calc
+import org.apache.calcite.rex.RexProgram
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.table.api.StreamQueryConfig
+import org.apache.flink.table.plan.schema.RowSchema
+import org.apache.flink.table.planner.StreamPlanner
+import org.apache.flink.table.runtime.types.CRow
+
+/**
+ * RelNode for Python ScalarFunctions.
+ */
+class DataStreamPythonCalc(
+cluster: RelOptCluster,
+traitSet: RelTraitSet,
+input: RelNode,
+inputSchema: RowSchema,
+schema: RowSchema,
+calcProgram: RexProgram,
+ruleDescription: String)
+ extends DataStreamCalcBase(
+cluster,
+traitSet,
+input,
+inputSchema,
+schema,
+calcProgram,
+ruleDescription) {
+
+ override def copy(traitSet: RelTraitSet, child: RelNode, program:
RexProgram): Calc = {
+new DataStreamPythonCalc(
+ cluster,
+ traitSet,
+ child,
+ inputSchema,
+ schema,
+ program,
+ ruleDescription)
+ }
+
+ override def translateToPlan(
+ planner: StreamPlanner,
+ queryConfig: StreamQueryConfig): DataStream[CRow] = {
+null
Review comment:
Add comment for this in this commit otherwise, other people would be
confused.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0 URL: https://github.com/apache/flink/pull/9712#discussion_r328414544 ## File path: flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/FlinkFilterJoinRuleTest.xml ## @@ -22,6 +22,7 @@ limitations under the License. SELECT * FROM (SELECT * FROM leftT WHERE a NOT IN (SELECT c FROM rightT WHERE c < 3)) T WHERE T.b > 2 ]]> + Review comment: Sure, thanks. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0 URL: https://github.com/apache/flink/pull/9712#discussion_r328414311 ## File path: flink-table/flink-table-planner-blink/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/DistinctAggregateTest.xml ## @@ -48,21 +48,23 @@ SortAggregate(isMerge=[true], select=[Final_COUNT(count$0) AS EXPR$0, Final_SUM(
[GitHub] [flink] danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
danny0405 commented on a change in pull request #9712: [FLINK-13656]
[sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to
1.21.0
URL: https://github.com/apache/flink/pull/9712#discussion_r328414214
##
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/calls/ScalarOperatorGens.scala
##
@@ -390,6 +390,16 @@ object ScalarOperatorGens {
}
}
+ def generateIsNotDistinctFrom(
+ ctx: CodeGeneratorContext,
+ left: GeneratedExpression,
+ right: GeneratedExpression)
+ : GeneratedExpression = {
Review comment:
Yep, thanks.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime] Reducing synchronization in AsyncWaitOperator
zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime]
Reducing synchronization in AsyncWaitOperator
URL: https://github.com/apache/flink/pull/9717#discussion_r328414259
##
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/execution/MailboxProcessor.java
##
@@ -200,7 +200,7 @@ private boolean processMail(TaskMailbox mailbox) throws
MailboxStateException, I
try {
maybeLetter.get().run();
} catch (Exception e) {
- e.printStackTrace();
+ throw new IllegalStateException("Cannot process
mail " + maybeLetter.get(), e);
Review comment:
From the description of `AsyncExceptionHandler`, it is proposed to used by
other threads except the main task thread. AFAIK it is only used by
`SplitReader` in `ContinuousFileReaderOperator` now. For this case if the
`processMail` is executed in the main task thread, maybe it is out of scope of
`AsyncExceptionHandler`.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0 URL: https://github.com/apache/flink/pull/9712#discussion_r328414087 ## File path: flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd ## @@ -394,23 +389,13 @@ ] # List of methods for parsing ddl supported data types. - # Return type of method implementation should be "SqlIdentifier". - # Example: SqlParseTimeStampZ(). - flinkDataTypeParserMethods: [ -"SqlArrayType()", -"SqlMultisetType()", -"SqlMapType()", -"SqlRowType()", -"SqlStringType()", -"SqlBytesType()", -"SqlTimestampType()", -"SqlTimeType()" - ] - - # List of methods for parsing custom data types. - # Return type of method implementation should be "SqlIdentifier". + # Return type of method implementation should be "SqlTypeNameSpec". # Example: SqlParseTimeStampZ(). dataTypeParserMethods: [ +"ExtendedSqlBasicTypeName()", +"CustomizedCollectionsTypeName()", Review comment: We already have a `ExtendedSqlCollectionsTypeName` parse block name, i name it `CustomizedXXX` because the syntax is totally customized and does not belong to standard SQL. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
danny0405 commented on a change in pull request #9712: [FLINK-13656]
[sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to
1.21.0
URL: https://github.com/apache/flink/pull/9712#discussion_r328413899
##
File path:
flink-table/flink-sql-parser/src/main/java/org/apache/calcite/sql/ExtendedSqlRowTypeNameSpec.java
##
@@ -0,0 +1,150 @@
+/*
+ * 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.calcite.sql;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.validate.SqlValidator;
+import org.apache.calcite.util.Litmus;
+import org.apache.calcite.util.Pair;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * A sql type name specification of ROW type.
+ *
+ * The difference with {@link SqlRowTypeNameSpec}:
+ *
+ * Support comment syntax for every field
+ * Field data type default is nullable
+ * Support ROW type with empty fields, e.g. ROW()
+ *
+ */
+public class ExtendedSqlRowTypeNameSpec extends SqlTypeNameSpec {
Review comment:
`SqlRowTypeNameSpec` has assertion that the row fields be must more than 1,
so i can not inherit from it, because we allow empty row: `ROW()`.
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0
danny0405 commented on a change in pull request #9712: [FLINK-13656] [sql-parser][table-planner][table-planner-blink] Bump Calcite dependency to 1.21.0 URL: https://github.com/apache/flink/pull/9712#discussion_r328413261 ## File path: flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl ## @@ -81,6 +66,33 @@ void TableColumn2(List list) : } } +/** +* Different with #DataType, we support a [ NULL | NOT NULL ] suffix syntax for both the Review comment: No, for `#DataType`, i mean the parser block name, maybe `#DataType()` is better. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-14080) Support precision of TimestampType
[ https://issues.apache.org/jira/browse/FLINK-14080?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Zhenghua Gao updated FLINK-14080: - Description: Timestamp values have a precision in fractional seconds that range from 0 to 9. For example, a precision of 0 means that no fractional seconds are stored, 3 means that the timestamp stores milliseconds, and 9 means a precision of nanoseconds. 0 is the minimum precision, and 9 is the maximum. Now, blink planner represent Timestamp values as Long(milliseconds) and only support precision is 3, we need support more precision Timestamp values. was:Now, blink planner only support precision is 3, we need support more because default precision is 6. > Support precision of TimestampType > -- > > Key: FLINK-14080 > URL: https://issues.apache.org/jira/browse/FLINK-14080 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / Planner >Reporter: Jingsong Lee >Priority: Major > Fix For: 1.10.0 > > > Timestamp values have a precision in fractional seconds that range from 0 to > 9. For example, a precision of 0 means that no fractional seconds are stored, > 3 means that the timestamp stores milliseconds, and 9 means a precision of > nanoseconds. 0 is the minimum precision, and 9 is the maximum. > Now, blink planner represent Timestamp values as Long(milliseconds) and only > support precision is 3, we need support more precision Timestamp values. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] wuchong commented on issue #9756: [backport to 1.9][FLINK-14145] Fix getLatestCheckpoint(true) returns wrong checkpoint
wuchong commented on issue #9756: [backport to 1.9][FLINK-14145] Fix getLatestCheckpoint(true) returns wrong checkpoint URL: https://github.com/apache/flink/pull/9756#issuecomment-535303241 Thanks @gyfora for merging this. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13740) TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
[
https://issues.apache.org/jira/browse/FLINK-13740?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938190#comment-16938190
]
Jark Wu commented on FLINK-13740:
-
Do you mean moving all the {{{materialize()}} logic to BinaryGenericSerializer?
I think the root problem is the BinaryGeneric object will be
serialize/deserialize/copy by multiple threads, but the
BinaryGeneric#copy()/materialize()/ensureMaterialized() uses the
{{javaObjectSer}} to (de)serializer which is not thread safe.
So I think even if we move copy logic to BinaryGenericSerializer.copy, there
still be problem in other methods.
> TableAggregateITCase.testNonkeyedFlatAggregate failed on Travis
> ---
>
> Key: FLINK-13740
> URL: https://issues.apache.org/jira/browse/FLINK-13740
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.9.0, 1.10.0
>Reporter: Till Rohrmann
>Priority: Critical
> Labels: test-stability
> Fix For: 1.10.0, 1.9.1
>
>
> The {{TableAggregateITCase.testNonkeyedFlatAggregate}} failed on Travis with
> {code}
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.table.planner.runtime.stream.table.TableAggregateITCase.testNonkeyedFlatAggregate(TableAggregateITCase.scala:93)
> Caused by: java.lang.Exception: Artificial Failure
> {code}
> https://api.travis-ci.com/v3/job/225551182/log.txt
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] AT-Fieldless edited a comment on issue #9718: [FLINK-13485] [chinese-translation] Translate "Table API Example Walkthrough" page into Chinese
AT-Fieldless edited a comment on issue #9718: [FLINK-13485] [chinese-translation] Translate "Table API Example Walkthrough" page into Chinese URL: https://github.com/apache/flink/pull/9718#issuecomment-533946231 @klion26 Do you mean using `git -rebase` to clear the git history like this? [https://stackoverflow.com/questions/30561895/removing-merge-commits-in-git?noredirect=1=1](url) 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13025) Elasticsearch 7.x support
[ https://issues.apache.org/jira/browse/FLINK-13025?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938180#comment-16938180 ] vinoyang commented on FLINK-13025: -- [~lilyevsky] Thanks for your verification and report your result. Maybe [~victorvilladev] want to verify sql connector es7 connector. Let's be patient to see if there is any problem. About the PR, I have addressed all the issues reported by [~aljoscha] and [~victorvilladev]. Now, the PR has passed the verification of Travis. Let's wait for another review. > Elasticsearch 7.x support > - > > Key: FLINK-13025 > URL: https://issues.apache.org/jira/browse/FLINK-13025 > Project: Flink > Issue Type: New Feature > Components: Connectors / ElasticSearch >Affects Versions: 1.8.0 >Reporter: Keegan Standifer >Priority: Major > Labels: pull-request-available > Attachments: flink-connector-elasticsearch7_2.12-1.10-SNAPSHOT.jar > > Time Spent: 10m > Remaining Estimate: 0h > > Elasticsearch 7.0.0 was released in April of 2019: > [https://www.elastic.co/blog/elasticsearch-7-0-0-released] > The latest elasticsearch connector is > [flink-connector-elasticsearch6|https://github.com/apache/flink/tree/master/flink-connectors/flink-connector-elasticsearch6] -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] dianfu edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
dianfu edited a comment on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-535291026 @sunjincheng121 Thanks for your comments. Very appreciated. The suggestions are very helpful and make sense to me. - Regarding non-parameter UDFs optimization and UDFs with constant parameters optimization, I have created https://issues.apache.org/jira/browse/FLINK-14212 and https://issues.apache.org/jira/browse/FLINK-14222 as a follow up. - Regarding to dependency of Beam, I will pay attention of it and will add documentation about it in https://issues.apache.org/jira/browse/FLINK-14027. - Regarding to dependency of cloud-pickle, I'm +1 to include it in the release of Flink. - Regarding to the interface of register_function, thanks a lot for your remind and I'll pay attention to FLIP-64. Will update the interface once it's accepted. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] dianfu commented on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work
dianfu commented on issue #9766: [FLINK-14018][python] Add Python building blocks to make sure the basic functionality of Python ScalarFunction could work URL: https://github.com/apache/flink/pull/9766#issuecomment-535291026 @sunjincheng121 Thanks for your comments. Very appreciated. The suggestions are very helpful and make sense to me. - Regarding non-parameter UDFs optimization and UDFs with constant parameters optimization, I have created https://issues.apache.org/jira/browse/FLINK-14212 and https://issues.apache.org/jira/browse/FLINK-14222 as a follow up. - Regarding to dependency of Beam, I will add some documentation about it in https://issues.apache.org/jira/browse/FLINK-14027. - Regarding to dependency of cloud-pickle, I'm +1 to include it in the release of Flink. - Regarding to the interface of register_function, thanks a lot for your remind and I'll pay attention to FLIP-64. Will update the interface once it's accepted. 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-14208) Optimize Python UDFs with parameters of constant values
[ https://issues.apache.org/jira/browse/FLINK-14208?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dian Fu updated FLINK-14208: Description: We need optimize Python UDFs with parameters of constant values. For example, the constant parameter isn't needed to be transferred between the Java operator and the Python worker. (was: We need provide native support for Python UDF with parameters of constant values. For example, the constant parameter isn't needed to be transferred between the Java operator and the Python worker.) > Optimize Python UDFs with parameters of constant values > --- > > Key: FLINK-14208 > URL: https://issues.apache.org/jira/browse/FLINK-14208 > Project: Flink > Issue Type: Sub-task > Components: API / Python >Reporter: Dian Fu >Priority: Major > Fix For: 1.10.0 > > > We need optimize Python UDFs with parameters of constant values. For example, > the constant parameter isn't needed to be transferred between the Java > operator and the Python worker. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14222) Optimize for Python UDFs with all parameters are constant values
Dian Fu created FLINK-14222: --- Summary: Optimize for Python UDFs with all parameters are constant values Key: FLINK-14222 URL: https://issues.apache.org/jira/browse/FLINK-14222 Project: Flink Issue Type: Sub-task Components: API / Python Reporter: Dian Fu Fix For: 1.10.0 As discussed in [https://github.com/apache/flink/pull/9766], The Python UDFs could be optimized to a constant value if it is deterministic. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-14208) Optimize Python UDFs with parameters of constant values
[ https://issues.apache.org/jira/browse/FLINK-14208?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Dian Fu updated FLINK-14208: Summary: Optimize Python UDFs with parameters of constant values (was: Support Python UDF with parameters of constant values) > Optimize Python UDFs with parameters of constant values > --- > > Key: FLINK-14208 > URL: https://issues.apache.org/jira/browse/FLINK-14208 > Project: Flink > Issue Type: Sub-task > Components: API / Python >Reporter: Dian Fu >Priority: Major > Fix For: 1.10.0 > > > We need provide native support for Python UDF with parameters of constant > values. For example, the constant parameter isn't needed to be transferred > between the Java operator and the Python worker. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-10672) Task stuck while writing output to flink
[
https://issues.apache.org/jira/browse/FLINK-10672?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16938147#comment-16938147
]
Kyle Weaver commented on FLINK-10672:
-
[~zjwang] I have tested the pipeline on Flink 1.8.1 and Flink 1.9.0 and the
issue is still present (even with BATCH_FORCED). Strangely, it didn't happen
when I had parallelism set to 11, but did happen when parallelism was 12.
However, it is solved if I increase the taskManager memory, as Ankur indicated
earlier.
> Task stuck while writing output to flink
>
>
> Key: FLINK-10672
> URL: https://issues.apache.org/jira/browse/FLINK-10672
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
>Affects Versions: 1.5.4
> Environment: OS: Debuan rodente 4.17
> Flink version: 1.5.4
> ||Key||Value||
> |jobmanager.heap.mb|1024|
> |jobmanager.rpc.address|localhost|
> |jobmanager.rpc.port|6123|
> |metrics.reporter.jmx.class|org.apache.flink.metrics.jmx.JMXReporter|
> |metrics.reporter.jmx.port|9250-9260|
> |metrics.reporters|jmx|
> |parallelism.default|1|
> |rest.port|8081|
> |taskmanager.heap.mb|1024|
> |taskmanager.numberOfTaskSlots|1|
> |web.tmpdir|/tmp/flink-web-bdb73d6c-5b9e-47b5-9ebf-eed0a7c82c26|
>
> h1. Overview
> ||Data Port||All Slots||Free Slots||CPU Cores||Physical Memory||JVM Heap
> Size||Flink Managed Memory||
> |43501|1|0|12|62.9 GB|922 MB|642 MB|
> h1. Memory
> h2. JVM (Heap/Non-Heap)
> ||Type||Committed||Used||Maximum||
> |Heap|922 MB|575 MB|922 MB|
> |Non-Heap|68.8 MB|64.3 MB|-1 B|
> |Total|991 MB|639 MB|922 MB|
> h2. Outside JVM
> ||Type||Count||Used||Capacity||
> |Direct|3,292|105 MB|105 MB|
> |Mapped|0|0 B|0 B|
> h1. Network
> h2. Memory Segments
> ||Type||Count||
> |Available|3,194|
> |Total|3,278|
> h1. Garbage Collection
> ||Collector||Count||Time||
> |G1_Young_Generation|13|336|
> |G1_Old_Generation|1|21|
>Reporter: Ankur Goenka
>Priority: Major
> Labels: beam
> Attachments: 1uruvakHxBu.png, 3aDKQ24WvKk.png, Po89UGDn58V.png,
> jmx_dump.json, jmx_dump_detailed.json, jstack_129827.log, jstack_163822.log,
> jstack_66985.log
>
>
> I am running a fairly complex pipleline with 200+ task.
> The pipeline works fine with small data (order of 10kb input) but gets stuck
> with a slightly larger data (300kb input).
>
> The task gets stuck while writing the output toFlink, more specifically it
> gets stuck while requesting memory segment in local buffer pool. The Task
> manager UI shows that it has enough memory and memory segments to work with.
> The relevant stack trace is
> {quote}"grpc-default-executor-0" #138 daemon prio=5 os_prio=0
> tid=0x7fedb0163800 nid=0x30b7f in Object.wait() [0x7fedb4f9]
> java.lang.Thread.State: TIMED_WAITING (on object monitor)
> at (C/C++) 0x7fef201c7dae (Unknown Source)
> at (C/C++) 0x7fef1f2aea07 (Unknown Source)
> at (C/C++) 0x7fef1f241cd3 (Unknown Source)
> at java.lang.Object.wait(Native Method)
> - waiting on <0xf6d56450> (a java.util.ArrayDeque)
> at
> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestMemorySegment(LocalBufferPool.java:247)
> - locked <0xf6d56450> (a java.util.ArrayDeque)
> at
> org.apache.flink.runtime.io.network.buffer.LocalBufferPool.requestBufferBuilderBlocking(LocalBufferPool.java:204)
> at
> org.apache.flink.runtime.io.network.api.writer.RecordWriter.requestNewBufferBuilder(RecordWriter.java:213)
> at
> org.apache.flink.runtime.io.network.api.writer.RecordWriter.sendToTarget(RecordWriter.java:144)
> at
> org.apache.flink.runtime.io.network.api.writer.RecordWriter.emit(RecordWriter.java:107)
> at
> org.apache.flink.runtime.operators.shipping.OutputCollector.collect(OutputCollector.java:65)
> at
> org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
> at
> org.apache.beam.runners.flink.translation.functions.FlinkExecutableStagePruningFunction.flatMap(FlinkExecutableStagePruningFunction.java:42)
> at
> org.apache.beam.runners.flink.translation.functions.FlinkExecutableStagePruningFunction.flatMap(FlinkExecutableStagePruningFunction.java:26)
> at
> org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:80)
> at
> org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
> at
> org.apache.beam.runners.flink.translation.functions.FlinkExecutableStageFunction$MyDataReceiver.accept(FlinkExecutableStageFunction.java:230)
> - locked <0xf6a60bd0> (a java.lang.Object)
> at
> org.apache.beam.sdk.fn.data.BeamFnDataInboundObserver.accept(BeamFnDataInboundObserver.java:81)
> at
> org.apache.beam.sdk.fn.data.BeamFnDataInboundObserver.accept(BeamFnDataInboundObserver.java:32)
> at
>
[jira] [Updated] (FLINK-14217) introduce temp functions to FunctionCatalog
[ https://issues.apache.org/jira/browse/FLINK-14217?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bowen Li updated FLINK-14217: - Summary: introduce temp functions to FunctionCatalog (was: introduce temp functions to replace "userFunctions" in FunctionCatalog) > introduce temp functions to FunctionCatalog > --- > > Key: FLINK-14217 > URL: https://issues.apache.org/jira/browse/FLINK-14217 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / API >Reporter: Bowen Li >Assignee: Bowen Li >Priority: Major > Fix For: 1.10.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-14216) introduce temp system functions to FunctionCatalog
[ https://issues.apache.org/jira/browse/FLINK-14216?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bowen Li updated FLINK-14216: - Summary: introduce temp system functions to FunctionCatalog (was: introduce temp system functions and APIs to FunctionCatalog) > introduce temp system functions to FunctionCatalog > -- > > Key: FLINK-14216 > URL: https://issues.apache.org/jira/browse/FLINK-14216 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / API >Reporter: Bowen Li >Assignee: Bowen Li >Priority: Major > Fix For: 1.10.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-14217) introduce temp functions to replace "userFunctions" in FunctionCatalog
[ https://issues.apache.org/jira/browse/FLINK-14217?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bowen Li updated FLINK-14217: - Summary: introduce temp functions to replace "userFunctions" in FunctionCatalog (was: introduce temp functions and APIS to replace "userFunctions" in FunctionCatalog) > introduce temp functions to replace "userFunctions" in FunctionCatalog > -- > > Key: FLINK-14217 > URL: https://issues.apache.org/jira/browse/FLINK-14217 > Project: Flink > Issue Type: Sub-task > Components: Table SQL / API >Reporter: Bowen Li >Assignee: Bowen Li >Priority: Major > Fix For: 1.10.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-14221) support drop and alter temp system functions
Bowen Li created FLINK-14221: Summary: support drop and alter temp system functions Key: FLINK-14221 URL: https://issues.apache.org/jira/browse/FLINK-14221 Project: Flink Issue Type: Sub-task Components: Table SQL / API Reporter: Bowen Li Assignee: Bowen Li Fix For: 1.10.0 -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-14220) support drop and alter temp functions
[ https://issues.apache.org/jira/browse/FLINK-14220?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bowen Li updated FLINK-14220: - Fix Version/s: 1.10.0 > support drop and alter temp functions > - > > Key: FLINK-14220 > URL: https://issues.apache.org/jira/browse/FLINK-14220 > Project: Flink > Issue Type: Sub-task >Reporter: Bowen Li >Priority: Major > Fix For: 1.10.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Assigned] (FLINK-14220) support drop and alter temp functions
[ https://issues.apache.org/jira/browse/FLINK-14220?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Bowen Li reassigned FLINK-14220: Assignee: Bowen Li > support drop and alter temp functions > - > > Key: FLINK-14220 > URL: https://issues.apache.org/jira/browse/FLINK-14220 > Project: Flink > Issue Type: Sub-task >Reporter: Bowen Li >Assignee: Bowen Li >Priority: Major > Fix For: 1.10.0 > > -- This message was sent by Atlassian Jira (v8.3.4#803005)
